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

.gitignore

@@ -7,5 +7,4 @@ target/
 logs/
 server/pki/
 client/src/main/python/setup.py
-grafana/timely-app/yarn-error.log
 

LICENSE

@@ -1,7 +1,7 @@
 
                                  Apache License
                            Version 2.0, January 2004
-                        httpProperties://www.apache.org/licenses/
+                        http://www.apache.org/licenses/
 
    TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
 
@@ -193,7 +193,7 @@
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at
 
-       httpProperties://www.apache.org/licenses/LICENSE-2.0
+       http://www.apache.org/licenses/LICENSE-2.0
 
    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,

README.md

@@ -2,7 +2,7 @@
 
 [![Apache License][li]][ll]
 
-Timely is a time series database application that provides secure access to time series data. Timely is written in Java and designed to work with [Apache Accumulo](https://accumuloProperties.apache.org/) and [Grafana](https://www.grafana.com). Documentation is located [here](https://nationalsecurityagency.github.io/timely/).
+Timely is a time series database application that provides secure access to time series data. Timely is written in Java and designed to work with [Apache Accumulo](https://accumulo.apache.org/) and [Grafana](https://www.grafana.com). Documentation is located [here](https://nationalsecurityagency.github.io/timely/).
 
 [li]: https://img.shields.io/badge/license-ASL-blue.svg
 [ll]: https://www.apache.org/licenses/LICENSE-2.0

analytics/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>1.2.8-SNAPSHOT</version>
   </parent>
   <artifactId>timely-analytics</artifactId>
   <!--
@@ -53,15 +53,16 @@
         </exclusion>
       </exclusions>
     </dependency>
-    <dependency>
-      <groupId>timely</groupId>
-      <artifactId>timely-common</artifactId>
-    </dependency>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>timely</groupId>
+      <artifactId>timely-common</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
   <build>
     <pluginManagement>
@@ -168,12 +169,6 @@
                 </excludes>
               </artifactSet>
               <filters>
-                <filter>
-                  <artifact>*:*</artifact>
-                  <excludes>
-                    <exclude>**/module-info.class</exclude>
-                  </excludes>
-                </filter>
                 <filter>
                   <artifact>org.apache.flink:*</artifact>
                   <excludes>
@@ -251,14 +246,6 @@
                   <artifactSet>
                     <excludes combine.self="override" />
                   </artifactSet>
-                  <filters>
-                    <filter>
-                      <artifact>*:*</artifact>
-                      <excludes>
-                        <exclude>**/module-info.class</exclude>
-                      </excludes>
-                    </filter>
-                  </filters>
                 </configuration>
               </execution>
             </executions>

analytics/src/main/java/timely/analytics/flink/MetricHistogram.java

@@ -7,7 +7,6 @@ import java.util.List;
 import java.util.stream.Collectors;
 
 import org.apache.flink.streaming.util.serialization.SerializationSchema;
-
 import timely.model.Tag;
 
 public class MetricHistogram implements Serializable, SerializationSchema<MetricHistogram> {

analytics/src/main/java/timely/analytics/flink/MetricKeySelector.java

@@ -1,13 +1,13 @@
 package timely.analytics.flink;
 
 import org.apache.flink.api.java.functions.KeySelector;
-
 import timely.api.response.MetricResponse;
 
 /**
- * Uniquely defines a time series in a Flink stream. A unique time series is the metric name and all of its tags, to include visibility if it exists.
+ * Uniquely defines a time series in a Flink stream. A unique time series is the
+ * metric name and all of its tags, to include visibility if it exists.
  */
-public class MetricKeySelector implements KeySelector<MetricResponse,String> {
+public class MetricKeySelector implements KeySelector<MetricResponse, String> {
 
     private static final long serialVersionUID = 1L;
 

analytics/src/main/java/timely/analytics/flink/SortedStringAccumulator.java

@@ -9,11 +9,11 @@ import org.apache.flink.api.common.accumulators.Accumulator;
  * Custom accumulator that retains a sorted map of strings and their counts
  *
  */
-public class SortedStringAccumulator implements Accumulator<String,ConcurrentSkipListMap<String,MutableLong>> {
+public class SortedStringAccumulator implements Accumulator<String, ConcurrentSkipListMap<String, MutableLong>> {
 
     private static final long serialVersionUID = 1L;
 
-    private ConcurrentSkipListMap<String,MutableLong> values = new ConcurrentSkipListMap<>();
+    private ConcurrentSkipListMap<String, MutableLong> values = new ConcurrentSkipListMap<>();
 
     @Override
     public void add(String value) {
@@ -25,7 +25,7 @@ public class SortedStringAccumulator implements Accumulator<String,ConcurrentSki
     }
 
     @Override
-    public ConcurrentSkipListMap<String,MutableLong> getLocalValue() {
+    public ConcurrentSkipListMap<String, MutableLong> getLocalValue() {
         return values;
     }
 
@@ -35,7 +35,7 @@ public class SortedStringAccumulator implements Accumulator<String,ConcurrentSki
     }
 
     @Override
-    public void merge(Accumulator<String,ConcurrentSkipListMap<String,MutableLong>> other) {
+    public void merge(Accumulator<String, ConcurrentSkipListMap<String, MutableLong>> other) {
         other.getLocalValue().forEach((k, v) -> {
             if (values.containsKey(k)) {
                 values.get(k).add(v.longValue());
@@ -46,7 +46,7 @@ public class SortedStringAccumulator implements Accumulator<String,ConcurrentSki
     }
 
     @Override
-    public Accumulator<String,ConcurrentSkipListMap<String,MutableLong>> clone() {
+    public Accumulator<String, ConcurrentSkipListMap<String, MutableLong>> clone() {
         SortedStringAccumulator a = new SortedStringAccumulator();
         a.getLocalValue().putAll(this.values);
         return a;

analytics/src/main/java/timely/analytics/flink/SubscriptionSource.java

@@ -22,7 +22,6 @@ import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
 import org.apache.flink.streaming.api.watermark.Watermark;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import timely.api.response.MetricResponse;
 import timely.api.response.MetricResponses;
 import timely.client.websocket.ClientHandler;
@@ -42,7 +41,7 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
         }
     }
 
-    private static final Logger log = LoggerFactory.getLogger(SubscriptionSource.class);
+    private static final Logger LOG = LoggerFactory.getLogger(SubscriptionSource.class);
     private static final long serialVersionUID = 1L;
 
     private transient WebSocketSubscriptionClient client = null;
@@ -64,11 +63,12 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
 
     @Override
     public void open(Configuration parameters) throws Exception {
-        log.info("Opening summarization job.");
+        LOG.info("Opening summarization job.");
         super.open(parameters);
-        client = new WebSocketSubscriptionClient(jp.getTimelyHostname(), jp.getTimelyHttpsPort(), jp.getTimelyWssPort(), jp.isClientAuth(), jp.isDoLogin(),
-                        jp.getKeyStoreFile(), jp.getKeyStoreType(), jp.getKeyStorePass(), jp.getTrustStoreFile(), jp.getTrustStoreType(),
-                        jp.getTrustStorePass(), jp.isHostVerificationEnabled(), jp.getBufferSize());
+        client = new WebSocketSubscriptionClient(jp.getTimelyHostname(), jp.getTimelyHttpsPort(), jp.getTimelyWssPort(),
+                jp.isClientAuth(), jp.isDoLogin(), jp.getTimelyUsername(), jp.getTimelyPassword(), jp.getKeyStoreFile(),
+                jp.getKeyStoreType(), jp.getKeyStorePass(), jp.getTrustStoreFile(), jp.getTrustStoreType(),
+                jp.getTrustStorePass(), jp.isHostVerificationEnabled(), jp.getBufferSize());
     }
 
     @Override
@@ -81,7 +81,7 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
 
     @Override
     public void run(SourceContext<MetricResponse> ctx) throws Exception {
-        log.info("Running summarization job");
+        LOG.info("Running summarization job");
         final SortedStringAccumulator dateTimeAccumulator = new SortedStringAccumulator();
         this.getRuntimeContext().addAccumulator("source hourly count", dateTimeAccumulator);
         final LongCounter sourceInputs = new LongCounter();
@@ -103,20 +103,22 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
 
                     @Override
                     public void onMessage(final String message) {
-                        log.info("Message received on Websocket session {}, length: {}", session.getId(), message.length());
+                        LOG.info("Message received on Websocket session {}, length: {}", session.getId(),
+                                message.length());
                         try {
                             // Deserialize in this thread
-                            final MetricResponses responses = JsonSerializer.getObjectMapper().readValue(message, MetricResponses.class);
+                            final MetricResponses responses = JsonSerializer.getObjectMapper().readValue(message,
+                                    MetricResponses.class);
                             // Process in the other
                             final Runnable r = () -> {
                                 responses.getResponses().forEach(response -> {
                                     if (response.isComplete()) {
-                                        log.info("Received last message.");
+                                        LOG.info("Received last message.");
                                         ctx.emitWatermark(new Watermark(Long.MAX_VALUE));
                                         subscriptionsRemaining.getAndDecrement();
                                         return;
                                     }
-                                    log.trace("Received metric: {}", response);
+                                    LOG.trace("Received metric: {}", response);
                                     long time = response.getTimestamp();
                                     ctx.collectWithTimestamp(response, time);
                                     dateTimeAccumulator.add(formatter.format(new Date(response.getTimestamp())));
@@ -139,11 +141,11 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
                                 inputProcessedInMainThread.add(1);
                             }
                         } catch (IOException e) {
-                            log.error("Error deserializing metric response, closing", e);
+                            LOG.error("Error deserializing metric response, closing", e);
                             try {
                                 close();
                             } catch (Exception e1) {
-                                log.error("Error closing client", e1);
+                                LOG.error("Error closing client", e1);
                             }
                         }
                     }
@@ -156,7 +158,7 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
                 try {
                     close();
                 } catch (Exception e1) {
-                    log.error("Error closing client", e1);
+                    LOG.error("Error closing client", e1);
                 }
                 // Signal done sending data
                 ctx.emitWatermark(new Watermark(Long.MAX_VALUE));
@@ -168,7 +170,7 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
                 try {
                     close();
                 } catch (Exception e1) {
-                    log.error("Error closing client", e1);
+                    LOG.error("Error closing client", e1);
                 }
             }
 
@@ -176,14 +178,14 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
         try {
             client.open(handler);
             for (String m : metrics) {
-                log.info("Adding subscription for {}", m);
+                LOG.info("Adding subscription for {}", m);
                 client.addSubscription(m, null, start, end, 5000);
             }
             while (!client.isClosed() && subscriptionsRemaining.get() > 0) {
                 try {
                     Thread.sleep(5000);
                 } catch (InterruptedException e) {
-                    log.error("Sleep interrupted.");
+                    LOG.error("Sleep interrupted.");
                 }
             }
         } finally {
@@ -193,21 +195,21 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
 
     @Override
     public void cancel() {
-        log.info("Cancelling subscription source.");
+        LOG.info("Cancelling subscription source.");
         try {
             close();
         } catch (Exception e) {
-            log.error("Error closing web socket client");
+            LOG.error("Error closing web socket client");
         }
     }
 
     @Override
     public void stop() {
-        log.info("Stopping subscription source.");
+        LOG.info("Stopping subscription source.");
         try {
             close();
         } catch (Exception e) {
-            log.error("Error closing web socket client");
+            LOG.error("Error closing web socket client");
         }
 
     }

analytics/src/main/java/timely/analytics/flink/SummarizationJob.java

@@ -13,12 +13,11 @@ import org.apache.flink.streaming.api.functions.sink.SocketClientSink;
 import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import timely.api.response.MetricResponse;
 
 public class SummarizationJob {
 
-    private static final Logger log = LoggerFactory.getLogger(SummarizationJob.class);
+    private static final Logger LOG = LoggerFactory.getLogger(SummarizationJob.class);
 
     public static void main(String[] args) throws Exception {
 
@@ -27,11 +26,11 @@ public class SummarizationJob {
             System.exit(1);
         }
 
-        log.info("Loading properties from: {}", args[0]);
+        LOG.info("Loading properties from: {}", args[0]);
         ParameterTool params = ParameterTool.fromPropertiesFile(args[0]);
 
         final SummarizationJobParameters jp = new SummarizationJobParameters(params);
-        log.info("Properties: {}", jp);
+        LOG.info("Properties: {}", jp);
 
         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
         // @formatter:off
@@ -43,7 +42,7 @@ public class SummarizationJob {
         .window(TumblingEventTimeWindows.of(jp.getSummarizationInterval()))
         .fold(new MetricHistogram(), new FoldFunction<MetricResponse, MetricHistogram>() {
 
-            private final Logger log = LoggerFactory
+            private final Logger LOG = LoggerFactory
                     .getLogger("timely.analytics.SummarizationJob.FoldFunction@");
             private static final long serialVersionUID = 1L;
 
@@ -59,7 +58,7 @@ public class SummarizationJob {
                         throw new RuntimeException("Tags don't not match - partitioning problem");
                     }
                 }
-                log.trace("Updating histogram for {} with value {} at {} for metric {}", histo.getMetric()
+                LOG.trace("Updating histogram for {} with value {} at {} for metric {}", histo.getMetric()
                         + histo.getTags(), metric.getValue(), metric.getTimestamp(), metric.getMetric()
                         + metric.getTags());
                 histo.update(metric.getValue(), metric.getTimestamp());
@@ -71,7 +70,7 @@ public class SummarizationJob {
             new SocketClientSink<MetricHistogram>(jp.getTimelyHostname(), jp.getTimelyTcpPort(),
                     new MetricHistogram(), 2, true) {
 
-                private final Logger log = LoggerFactory
+                private final Logger LOG = LoggerFactory
                         .getLogger("timely.analytics.SummarizationJob.SocketClientSink");
                 private static final long serialVersionUID = 1L;
                 private final LongCounter sinkCounter = new LongCounter();
@@ -83,7 +82,7 @@ public class SummarizationJob {
                     super.open(parameters);
                     this.getRuntimeContext().addAccumulator("sink counter", this.sinkCounter);
                     this.getRuntimeContext().addAccumulator("sink hourly count", dateTimeAccumulator);
-                    log.info("Sink opened");
+                    LOG.info("Sink opened");
                 }
 
                 @Override
@@ -97,7 +96,7 @@ public class SummarizationJob {
                 @Override
                 public void close() throws Exception {
                     super.close();
-                    log.info("Sink closed");
+                    LOG.info("Sink closed");
                 }
 
             }).name("Timely Metric Sink");

analytics/src/main/java/timely/analytics/flink/SummarizationJobParameters.java

@@ -184,8 +184,8 @@ public class SummarizationJobParameters extends GlobalJobParameters implements S
     }
 
     @Override
-    public Map<String,String> toMap() {
-        Map<String,String> m = new HashMap<>();
+    public Map<String, String> toMap() {
+        Map<String, String> m = new HashMap<>();
         m.put("timelyHostname", this.timelyHostname);
         m.put("timelyHttpsPort", Integer.toString(this.timelyHttpsPort));
         m.put("timelyWssPort", Integer.toString(this.timelyWssPort));

analytics/src/main/spotbugs/excludes.xml

@@ -15,12 +15,4 @@
     <Method name="getMetrics" />
     <Bug pattern="EI_EXPOSE_REP" />
   </Match>
-  <Match>
-    <!-- these are probably avoidable, but needs further triage to fix -->
-    <Bug pattern="EI_EXPOSE_REP" />
-  </Match>
-  <Match>
-    <!-- these are probably avoidable, but needs further triage to fix -->
-    <Bug pattern="EI_EXPOSE_REP2" />
-  </Match>
 </FindBugsFilter>

analytics/src/test/java/timely/analytics/flink/MetricHistogramTest.java

@@ -6,7 +6,6 @@ import java.util.List;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
-
 import timely.model.Metric;
 import timely.model.Tag;
 import timely.model.parse.MetricParser;
@@ -97,8 +96,9 @@ public class MetricHistogramTest {
             Metric metric = metricParser.parse(put);
             Assert.assertEquals("sys.cpu.user_summarized", metric.getName());
             metric.getTags().forEach(t -> {
-                Assert.assertTrue(t.equals(t1) || t.equals(t2) || t.equals(avg) || t.equals(min) || t.equals(max) || t.equals(sum) || t.equals(count)
-                                || t.equals(p50) || t.equals(p75) || t.equals(p90) || t.equals(p99));
+                Assert.assertTrue(
+                        t.equals(t1) || t.equals(t2) || t.equals(avg) || t.equals(min) || t.equals(max) || t.equals(sum)
+                                || t.equals(count) || t.equals(p50) || t.equals(p75) || t.equals(p90) || t.equals(p99));
             });
         }
     }

balancer/bin/create-timely-balancer-profiles.sh

@@ -1,40 +0,0 @@
-#!/bin/bash
-
-if [[ `uname` == "Darwin" ]]; then
-    THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
-else
-    THIS_SCRIPT=`readlink -f $0`
-fi
-
-THIS_DIR="${THIS_SCRIPT%/*}"
-BASE_DIR=$(cd $THIS_DIR/.. && pwd)
-BIN_DIR=${BASE_DIR}/bin
-CONF_DIR=${BASE_DIR}/conf
-
-. ${BIN_DIR}/timely-balancer-env.sh
-
-for i in $(seq 1 $NUM_PROFILES); do
-    if [ "$NUM_PROFILES" -eq "1" ]; then
-        FILENAME=timely-balancer.yml
-    else
-        FILENAME=timely-balancer-${i}.yml
-    fi
-
-    echo "creating configuration file ${CONF_DIR}/${FILENAME}"
-    cp ${CONF_DIR}/timely-balancer-template.yml ${CONF_DIR}/${FILENAME}
-#    if [ "$NUM_PROFILES" -eq "1" ]; then
-#        sed -i "s/\${INSTANCE}//g" ${CONF_DIR}/${FILENAME}
-#    else
-#        sed -i "s/\${INSTANCE}/${i}/g" ${CONF_DIR}/${FILENAME}
-#    fi
-    MULTIPLE=$(($i - 1))
-    TCP_PORT=$(($TCP_PORT_START + $MULTIPLE * $PORT_INCREMENT))
-    sed -i "s/\${TCP_PORT}/${TCP_PORT}/g" ${CONF_DIR}/${FILENAME}
-    HTTP_PORT=$(($HTTP_PORT_START + $MULTIPLE * $PORT_INCREMENT))
-    sed -i "s/\${HTTP_PORT}/${HTTP_PORT}/g" ${CONF_DIR}/${FILENAME}
-    WEBSOCKET_PORT=$(($WEBSOCKET_PORT_START + $MULTIPLE * $PORT_INCREMENT))
-    sed -i "s/\${WEBSOCKET_PORT}/${WEBSOCKET_PORT}/g" ${CONF_DIR}/${FILENAME}
-    UDP_PORT=$(($UDP_PORT_START + $MULTIPLE * $PORT_INCREMENT))
-    sed -i "s/\${UDP_PORT}/${UDP_PORT}/g" ${CONF_DIR}/${FILENAME}
-done
-

balancer/bin/start-all.sh

@@ -1,23 +0,0 @@
-#!/bin/bash
-
-if [[ `uname` == "Darwin" ]]; then
-    THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
-else
-    THIS_SCRIPT=`readlink -f $0`
-fi
-
-THIS_DIR="${THIS_SCRIPT%/*}"
-BASE_DIR=$(cd $THIS_DIR/.. && pwd)
-BIN_DIR=${BASE_DIR}/bin
-
-. ${BIN_DIR}/timely-balancer-env.sh
-
-if [ "$NUM_PROFILES" -eq "1" ]; then
-    echo ${BIN_DIR}/timely-balancer.sh
-    ${BIN_DIR}/timely-balancer.sh
-else
-    for i in $(seq 1 $NUM_PROFILES); do
-        echo ${BIN_DIR}/timely-balancer.sh $i
-        ${BIN_DIR}/timely-balancer.sh $i
-    done
-fi

balancer/bin/stop-all.sh

@@ -1,3 +0,0 @@
-#!/bin/bash
-
-pkill -9 -f 'timely-balancer-.*-exec.jar'

balancer/bin/timely-balancer-env.sh

@@ -1,27 +0,0 @@
-#!/bin/bash
-
-NUM_PROFILES=1
-ACCUMULO_ZOOKEEPERS=localhost:2181
-TCP_PORT_START=4242
-HTTP_PORT_START=4243
-WEBSOCKET_PORT_START=4244
-UDP_PORT_START=4245
-PORT_INCREMENT=100
-
-HOST_IP=0.0.0.0
-HOST=$(hostname -s)
-HOST_FQDN=$(hostname -f)
-
-GENERATE_KEYS=true
-TRUST_STORE_TYPE=
-TRUST_STORE_FILE=
-TRUST_STORE_PASSWORD=
-KEY_STORE_TYPE=
-KEY_STORE_FILE=
-KEY_STORE_PASSWORD=
-AUTHORIZATION_URL=
-AUTHORIZATION_CERTIFICATE_TYPE=
-AUTHORIZATION_CERTIFICATE_FILE=
-
-REQUIRED_ROLES=
-REQUIRED_AUTHS=

balancer/bin/timely-balancer.sh

@@ -9,61 +9,34 @@ else
 fi
 
 THIS_DIR="${THIS_SCRIPT%/*}"
+NATIVE_DIR="${THIS_DIR}/META-INF/native"
+BASE_DIR=${THIS_DIR}/..
+TMP_DIR="${BASE_DIR}/tmp"
+CONF_DIR="${BASE_DIR}/conf"
+LIB_DIR="${BASE_DIR}/lib"
+NUM_SERVER_THREADS=4
 
-set -a
-. ${THIS_DIR}/timely-balancer-env.sh
-set +a
-
-if [ -n "$1" ]; then
-  export INSTANCE="$1"
-  PROFILE_ARG="--spring.profiles.active=${INSTANCE}"
-  PID=`jps -m | grep -E 'timely-balancer-.*-exec' | grep "spring.profiles.active=${INSTANCE}" | awk '{print $1}'`
-else
-  export INSTANCE="1"
-  PID=`jps -m | grep -E 'timely-balancer-.*-exec' | awk '{print $1}'`
+if [[ -e ${TMP_DIR} ]]; then
+  rm -rf ${TMP_DIR}
 fi
+mkdir ${TMP_DIR}
 
-if [ "$PID" == "" ]; then
-
-    NATIVE_DIR="${THIS_DIR}/META-INF/native"
-    BASE_DIR=$(cd $THIS_DIR/.. && pwd)
-    TMP_DIR="${BASE_DIR}/tmp"
-    export CONF_DIR="${BASE_DIR}/conf"
-    LIB_DIR="${BASE_DIR}/lib"
-    export LOG_DIR="${BASE_DIR}/logs"
-    NUM_SERVER_THREADS=4
-
-    if [[ -e ${TMP_DIR} ]]; then
-      rm -rf ${TMP_DIR}
-    fi
-    mkdir ${TMP_DIR}
-
-    if [[ -e ${NATIVE_DIR} ]]; then
-      rm -rf ${NATIVE_DIR}
-    fi
-    mkdir -p ${NATIVE_DIR}
-
-    if [[ ! -e ${LOG_DIR} ]]; then
-      mkdir ${LOG_DIR}
-    fi
-
-    pushd ${BASE_DIR}/bin
-    $JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative-boringssl-static*.jar META-INF/native/libnetty_tcnative_linux_x86_64.${TCNATIVE_SUFFIX}
-    $JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-all*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
-    popd
-
-    export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*:${HADOOP_CONF_DIR}"
-    JVM_ARGS="-Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} -Dlog4j.configurationFile=${CONF_DIR}/log4j2.yml"
-    JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
-    JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
-
-    JVM_ARGS="${JVM_ARGS} -Xmx2G -Xms2G -XX:NewSize=1G -XX:MaxNewSize=1G"
-    JVM_ARGS="${JVM_ARGS} -XX:+UseCompressedOops -XX:+UseNUMA -Djava.net.preferIPv4Stack=true"
-    JVM_ARGS="${JVM_ARGS} -XX:+UseG1GC -XX:MaxGCPauseMillis=4000 -XX:ParallelGCThreads=20 -XX:+UseStringDeduplication"
-
-    echo "$JAVA_HOME/bin/java ${JVM_ARGS} -jar ${THIS_DIR}/timely-balancer-*-exec.jar ${PROFILE_ARG}"
-    nohup $JAVA_HOME/bin/java ${JVM_ARGS} -jar ${THIS_DIR}/timely-balancer-*-exec.jar ${PROFILE_ARG} >> ${LOG_DIR}/timely-balancer${INSTANCE}.out 2>&1 &
-
-else
-    echo "timely-balancer ${INSTANCE} already running with pid ${PID}"
+if [[ -e ${NATIVE_DIR} ]]; then
+  rm -rf ${NATIVE_DIR}
 fi
+mkdir -p ${NATIVE_DIR}
+
+pushd ${BASE_DIR}/bin
+$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative-boringssl-static*.jar META-INF/native/libnetty_tcnative_linux_x86_64.${TCNATIVE_SUFFIX}
+$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-all*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
+popd
+
+export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*:${HADOOP_CONF_DIR}"
+JVM_ARGS="-Xmx1G -Xms1G -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} -Dlog4j.configurationFile=${CONF_DIR}/log4j2-spring.xml"
+JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
+JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
+
+#JVM_ARGS="${JVM_ARGS} -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=5005"
+
+echo "$JAVA_HOME/bin/java ${JVM_ARGS} timely.balancer.Balancer --spring.config.name=timely --spring.profiles.active=balancer"
+$JAVA_HOME/bin/java ${JVM_ARGS} timely.balancer.Balancer --spring.config.name=timely --spring.profiles.active=balancer

balancer/conf/log4j2-error-console.xml

@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="warn" monitorInterval="60">
+  <Appenders>
+    <Console name="Console" target="SYSTEM_OUT">
+      <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
+    </Console>
+  </Appenders>
+  <Loggers>
+    <Root level="error" includeLocation="false">
+      <AppenderRef ref="Console" level="error" />
+    </Root>
+  </Loggers>
+</Configuration>

balancer/conf/log4j2-spring.xml

@@ -0,0 +1,29 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="warn" monitorInterval="60">
+  <Properties>
+    <Property name="PID">????</Property>
+    <Property name="LOG_PATTERN">%clr{%d{yyyy-MM-dd HH:mm:ss.SSS}}{faint} %clr{%5p} %clr{${sys:PID}}{magenta} %clr{---}{faint} %clr{[%15.15t]}{faint} %clr{%-40.40c{1.}}{cyan} %clr{:}{faint} %m%n%wEx</Property>
+  </Properties>
+  <Appenders>
+    <RollingRandomAccessFile name="Rolling" fileName="/srv/logs/timely/timely-balancer.log"
+    						 filePattern="/srv/logs/timely/$${date:yyyy-MM}/timely-balancer-%d{MM-dd-yyyy}-%i.log.gz">
+      <PatternLayout pattern="${LOG_PATTERN}"/>
+      <Policies>
+      	<SizeBasedTriggeringPolicy size="1024 MB" />
+      </Policies>
+      <DefaultRolloverStrategy max="10" />
+    </RollingRandomAccessFile>
+    <Console name="Console" target="SYSTEM_OUT">
+      <PatternLayout pattern="${LOG_PATTERN}"/>
+    </Console>
+  </Appenders>
+  <Loggers>
+    <Root level="trace" includeLocation="false">
+      <AppenderRef ref="Console" level="warn" />
+      <AppenderRef ref="Rolling" level="trace" />
+    </Root>
+    <Logger name="timely" additivity="false">
+      <AppenderRef ref="Console" level="debug" />
+    </Logger>
+  </Loggers>
+</Configuration>

balancer/conf/log4j2.yml

@@ -1,43 +0,0 @@
-Configuration:
-  status: warn
-  monitorInterval: 60
-
-  Properties:
-    Property:
-      - name: logDir
-        value: ${env:LOG_DIR}
-      - name: PID
-        value: "????"
-      - name: LOG_PATTERN
-        value: "%clr{%d{yyyy-MM-dd HH:mm:ss.SSS}}{faint} %clr{%5p} %clr{${sys:PID}}{magenta} %clr{---}{faint} %clr{[%15.15t]}{faint} %clr{%-40.40c{1.}}{cyan} %clr{:}{faint} %m%n%wEx"
-
-  Appenders:
-    Console:
-      name: Console
-      target: SYSTEM_OUT
-      follow: true
-      PatternLayout:
-        pattern: "${LOG_PATTERN}"
-
-    RollingFile:
-      - name: File
-        fileName: "${sys:logDir}/timely-balancer${env:INSTANCE}.log"
-        filePattern: "${sys:logDir}/timely-balancer${env:INSTANCE}.log.%d{yyyy-MM-dd}-%i.gz"
-        append: true
-        bufferedIO: true
-        bufferSize: 8192
-        Policies:
-          TimeBasedTriggeringPolicy:
-            interval: 1
-          SizeBasedTriggeringPolicy:
-            size: 250MB
-        DefaultRolloverStrategy:
-          max: 10
-        PatternLayout:
-          pattern: "${LOG_PATTERN}"
-
-  Loggers:
-    Root:
-      level: info
-      AppenderRef:
-        - ref: File

balancer/conf/security.properties

@@ -1,9 +0,0 @@
-trust.store.type=${TRUST_STORE_TYPE}
-trust.store.file=${TRUST_STORE_FILE}
-trust.store.password=${TRUST_STORE_PASSWORD}
-key.store.type=${KEY_STORE_TYPE}
-key.store.file=${KEY_STORE_FILE}
-key.store.password=${KEY_STORE_PASSWORD}
-authorization.url=${AUTHORIZATION_URL}
-authorization.certificate.type=${AUTHORIZATION_CERTIFICATE_TYPE}
-authorization.certificate.file=${AUTHORIZATION_CERTIFICATE_FILE}

balancer/conf/security.xml

@@ -0,0 +1,71 @@
+<beans:beans xmlns="http://www.springframework.org/schema/security"
+             xmlns:beans="http://www.springframework.org/schema/beans"
+             xmlns:util="http://www.springframework.org/schema/util"
+             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+             xsi:schemaLocation="http://www.springframework.org/schema/beans
+								 http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
+								 http://www.springframework.org/schema/security
+								 http://www.springframework.org/schema/security/spring-security.xsd
+								 http://www.springframework.org/schema/util
+								 http://www.springframework.org/schema/util/spring-util.xsd">
+		          
+	<beans:bean id="x509PrincipalExtractor" class="org.springframework.security.web.authentication.preauth.x509.SubjectDnX509PrincipalExtractor">
+		<beans:property name="subjectDnRegex" value="CN=(.*?)(?:,|$)"/>
+	</beans:bean>
+
+	<beans:bean id="timelyUserDetailsService" class="timely.auth.UserDetailsService">
+		<beans:property name="users">
+			<util:list>
+				<beans:bean class="timely.auth.TimelyUser">
+					<beans:constructor-arg name="subjectDn">
+						<beans:value>CN=example.com</beans:value>
+					</beans:constructor-arg>
+					<beans:constructor-arg name="auths">
+						<util:list>
+							<beans:value>D</beans:value>
+							<beans:value>E</beans:value>
+							<beans:value>F</beans:value>
+						</util:list>
+					</beans:constructor-arg>
+					<beans:constructor-arg name="roles">
+						<util:list>
+							<beans:value>G</beans:value>
+							<beans:value>H</beans:value>
+							<beans:value>I</beans:value>
+						</util:list>
+					</beans:constructor-arg>
+				</beans:bean>
+			</util:list>
+		</beans:property>
+	</beans:bean>
+    
+    <beans:bean id="x509AuthorizationProvider" class="org.springframework.security.web.authentication.preauth.PreAuthenticatedAuthenticationProvider">
+    	<beans:property name="preAuthenticatedUserDetailsService" ref="timelyUserDetailsService" />
+    </beans:bean>
+    
+    <user-service id="usernamePasswordService">
+    	<user name="test" password="test1" authorities="A,B,C" />
+    </user-service>
+
+	<beans:bean id="noOpPasswordEncoder" class="org.springframework.security.crypto.password.NoOpPasswordEncoder"/>
+
+	<beans:bean id="UserNamePasswordAuthenticationProvider" class="org.springframework.security.authentication.dao.DaoAuthenticationProvider">
+		<beans:property name="userDetailsService" ref="usernamePasswordService"/>
+		<beans:property name="passwordEncoder" ref="noOpPasswordEncoder"/>
+	</beans:bean>
+	    
+	<authentication-manager alias="authenticationManager">
+		<!-- X509 PreAuthenticated authorization provider -->
+		<authentication-provider ref="x509AuthorizationProvider"/>
+		<!-- Basic authorization provider, validates provided username and password -->
+		<authentication-provider ref="UserNamePasswordAuthenticationProvider" />
+	</authentication-manager>
+
+	<util:list id="requiredRoles" value-type="java.lang.String">
+		<beans:value></beans:value>
+	</util:list>
+
+	<util:list id="requiredAuths" value-type="java.lang.String">
+		<beans:value></beans:value>
+	</util:list>
+</beans:beans>

balancer/conf/timely-balancer-template.yml

@@ -1,142 +0,0 @@
-timely:
-  zookeeper:
-    servers: ${ACCUMULO_ZOOKEEPERS}
-    timeout: 30s
-  security:
-    session-max-age: 86400
-    cache-expiration-minutes: 60
-    cache-refresh-minutes: 5
-    allow-anonymous-http-access: true
-    allow-anonymous-ws-access: true
-    serverSsl:
-      key-store-type: ${KEY_STORE_TYPE}
-      key-store-file: ${KEY_STORE_FILE}
-      key-store-password: ${KEY_STORE_PASSWORD}
-      use-generated-keypair: ${GENERATE_KEYS}
-      ignore-ssl-handshake-errors: true
-      trust-store-type: ${TRUST_STORE_TYPE}
-      trust-store-file: ${TRUST_STORE_FILE}
-      trust-store-password: ${TRUST_STORE_PASSWORD}
-    clientSsl:
-      key-store-type: ${KEY_STORE_TYPE}
-      key-store-file: ${KEY_STORE_FILE}
-      key-store-password: ${KEY_STORE_PASSWORD}
-      trust-store-type: ${TRUST_STORE_TYPE}
-      trust-store-file: ${TRUST_STORE_FILE}
-      trust-store-password: ${TRUST_STORE_PASSWORD}
-    jwtSsl:
-      key-store-type: ${AUTHORIZATION_CERTIFICATE_TYPE}
-      key-store-file: ${AUTHORIZATION_CERTIFICATE_FILE}
-      key-store-password: ${AUTHORIZATION_CERTIFICATE_PASSWORD}
-  cache:
-    enabled: false
-    metric-age-off-hours:
-      default: 26
-    nonCachedMetrics:
-  balancer:
-    checkServerHealthInterval: 10000
-    serverFailuresBeforeDown: 6
-    serverSuccessesBeforeUp: 1
-    assignmentFile: /timely/balancer/assignments.csv
-    server:
-      ip: ${HOST_IP}
-      tcpPort: ${TCP_PORT}
-      numTcpPools: 1
-      tcpBufferSize: 100
-      tcpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-      udpPort: ${UDP_PORT}
-      udpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-    http:
-      host: ${HOST}
-      ip: ${HOST_IP}
-      port: ${HTTP_PORT}
-      redirect-path: /secure-me
-      strict-transport-max-age: 604800
-      cors:
-        allow-any-origin: false
-        allow-null-origin: false
-        allow-credentials: true
-        allowed-origins:
-        allowed-methods: DELETE,GET,HEAD,OPTIONS,PUT,POST
-        allowed-headers: content-type
-      httpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-    websocket:
-      ip: ${HOST_IP}
-      port: ${WEBSOCKET_PORT}
-      timeout: 60
-      subscription-lag: 120
-      scanner-batch-size: 10000
-      flush-interval-seconds: 30
-      scanner-read-ahead: 1
-      subscription-batch-size: 5000
-      incomingBufferSize: 50000000
-      wsClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-
-logging:
-  level:
-    ROOT: info
-    timey: debug

balancer/conf/timely-balancer.yml

@@ -1,7 +1,11 @@
-timely:
-  zookeeper:
-    servers: ${ACCUMULO_ZOOKEEPERS}
+timely-balancer:
+  zooKeeper:
+    servers: localhost:2181
     timeout: 30s
+  checkServerHealthInterval: 10000
+  serverFailuresBeforeDown: 6
+  serverSuccessesBeforeUp: 1
+  assignmentFile: /timely/balancer/assignments.csv
   security:
     session-max-age: 86400
     cache-expiration-minutes: 60
@@ -9,134 +13,119 @@ timely:
     allow-anonymous-http-access: true
     allow-anonymous-ws-access: true
     serverSsl:
-      key-store-type:
-      key-store-file:
-      key-store-password:
+      certificate-file:
+      key-file:
+      #key-password:
       use-generated-keypair: true
-      ignore-ssl-handshake-errors: true
-      trust-store-type:
       trust-store-file:
-      trust-store-password:
+      use-openssl: true
+      #use-ciphers:
     clientSsl:
-      key-store-type:
-      key-store-file:
-      key-store-password:
-      trust-store-type:
-      trust-store-file:
-      trust-store-password:
-    jwtSsl:
-      key-store-type:
-      key-store-file:
-      key-store-password:
+      keyFile:
+      keyType: pkcs12
+      #keyPassword:
+      trustStoreFile:
+      trustStoreType: jks
+      #trustStorePassword:
   cache:
     enabled: false
     metric-age-off-hours:
       default: 26
     nonCachedMetrics:
-  balancer:
-    checkServerHealthInterval: 10000
-    serverFailuresBeforeDown: 6
-    serverSuccessesBeforeUp: 1
-    assignmentFile: /timely/balancer/assignments.csv
-    server:
-      ip: 0.0.0.0
-      tcpPort: 4242
-      numTcpPools: 1
-      tcpBufferSize: 100
-      tcpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-      udpPort: 4245
-      udpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-    http:
-      host: localhost
-      ip: 0.0.0.0
-      port: 4243
-      redirect-path: /secure-me
-      strict-transport-max-age: 604800
-      cors:
-        allow-any-origin: false
-        allow-null-origin: false
-        allow-credentials: true
-        allowed-origins:
-        allowed-methods: DELETE,GET,HEAD,OPTIONS,PUT,POST
-        allowed-headers: content-type
-      httpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-    websocket:
-      ip: 0.0.0.0
-      port: 5244
-      timeout: 60
-      subscription-lag: 120
-      scanner-batch-size: 10000
-      flush-interval-seconds: 30
-      scanner-read-ahead: 1
-      subscription-batch-size: 5000
-      incomingBufferSize: 50000000
-      wsClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-
-logging:
-  level:
-    ROOT: info
-    timey: debug
+  server:
+    ip: 127.0.0.1
+    tcpPort: 4242
+    numTcpPools: 1
+    tcpBufferSize: 100
+    tcpClientPool:
+      minIdlePerKey: 0
+      maxIdlePerKey: 40
+      maxTotalPerKey: 40
+      maxTotal: -1
+      lifo: true
+      fairness: false
+      maxWaitMillis: -1
+      minEvictableIdleTimeMillis: 1800000
+      evictorShutdownTimeoutMillis: 10000
+      softMinEvictableIdleTimeMillis: -1
+      numTestsPerEvictionRun: 3
+      timeBetweenEvictionRunsMillis: -1
+      blockWhenExhausted: true
+      jmxEnabled: true
+      testOnBorrow: false
+      testOnReturn: false
+    udpPort: 4245
+    udpClientPool:
+      minIdlePerKey: 0
+      maxIdlePerKey: 40
+      maxTotalPerKey: 40
+      maxTotal: -1
+      lifo: true
+      fairness: false
+      maxWaitMillis: -1
+      minEvictableIdleTimeMillis: 1800000
+      evictorShutdownTimeoutMillis: 10000
+      softMinEvictableIdleTimeMillis: -1
+      numTestsPerEvictionRun: 3
+      timeBetweenEvictionRunsMillis: -1
+      blockWhenExhausted: true
+      jmxEnabled: true
+      testOnBorrow: false
+      testOnReturn: false
+  http:
+    host: localhost
+    ip: 127.0.0.1
+    port: 4243
+    redirect-path: /secure-me
+    strict-transport-max-age: 608400
+    cors:
+      allow-any-origin: false
+      allow-null-origin: false
+      allow-credentials: true
+      allowed-origins:
+      allowed-methods: DELETE,GET,HEAD,OPTIONS,PUT,POST
+      allowed-headers: content-type
+    httpClientPool:
+      minIdlePerKey: 0
+      maxIdlePerKey: 40
+      maxTotalPerKey: 40
+      maxTotal: -1
+      lifo: true
+      fairness: false
+      maxWaitMillis: -1
+      minEvictableIdleTimeMillis: 1800000
+      evictorShutdownTimeoutMillis: 10000
+      softMinEvictableIdleTimeMillis: -1
+      numTestsPerEvictionRun: 3
+      timeBetweenEvictionRunsMillis: -1
+      blockWhenExhausted: true
+      jmxEnabled: true
+      testOnBorrow: false
+      testOnReturn: false
+  websocket:
+    ip: 127.0.0.1
+    port: 5244
+    timeout: 60
+    subscription-lag: 120
+    scanner-batch-size: 10000
+    flush-interval-seconds: 30
+    scanner-read-ahead: 1
+    subscription-batch-size: 5000
+    incomingBufferSize: 50000000
+    wsClientPool:
+      minIdlePerKey: 0
+      maxIdlePerKey: 40
+      maxTotalPerKey: 40
+      maxTotal: -1
+      lifo: true
+      fairness: false
+      maxWaitMillis: -1
+      minEvictableIdleTimeMillis: 1800000
+      evictorShutdownTimeoutMillis: 10000
+      softMinEvictableIdleTimeMillis: -1
+      numTestsPerEvictionRun: 3
+      timeBetweenEvictionRunsMillis: -1
+      blockWhenExhausted: true
+      jmxEnabled: true
+      testOnBorrow: false
+      testOnReturn: false

balancer/pom.xml

@@ -4,13 +4,9 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>1.2.8-SNAPSHOT</version>
   </parent>
   <artifactId>timely-balancer</artifactId>
-  <description>Timely Balancer</description>
-  <properties>
-    <start-class>timely.balancer.TimelyBalancer</start-class>
-  </properties>
   <dependencies>
     <dependency>
       <groupId>com.fasterxml.jackson.core</groupId>
@@ -41,12 +37,8 @@
       <artifactId>commons-beanutils</artifactId>
     </dependency>
     <dependency>
-      <groupId>commons-logging</groupId>
-      <artifactId>commons-logging</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>io.github.classgraph</groupId>
-      <artifactId>classgraph</artifactId>
+      <groupId>io.github.lukehutch</groupId>
+      <artifactId>fast-classpath-scanner</artifactId>
     </dependency>
     <dependency>
       <groupId>io.netty</groupId>
@@ -59,6 +51,12 @@
     <dependency>
       <groupId>org.apache.accumulo</groupId>
       <artifactId>accumulo-core</artifactId>
+      <exclusions>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-log4j12</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
       <groupId>org.apache.accumulo</groupId>
@@ -99,23 +97,25 @@
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-common</artifactId>
+      <exclusions>
+        <exclusion>
+          <groupId>jdk.tools</groupId>
+          <artifactId>jdk.tools</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-minicluster</artifactId>
     </dependency>
     <dependency>
-      <groupId>org.apache.logging.log4j</groupId>
-      <artifactId>log4j-1.2-api</artifactId>
+      <groupId>org.glassfish.tyrus</groupId>
+      <artifactId>tyrus-client</artifactId>
     </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>
     </dependency>
-    <dependency>
-      <groupId>org.springframework</groupId>
-      <artifactId>spring-webflux</artifactId>
-    </dependency>
     <dependency>
       <groupId>org.springframework</groupId>
       <artifactId>spring-websocket</artifactId>
@@ -138,18 +138,6 @@
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-validation</artifactId>
     </dependency>
-    <dependency>
-      <groupId>org.springframework.boot</groupId>
-      <artifactId>spring-boot-starter-web</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.springframework.cloud</groupId>
-      <artifactId>spring-cloud-context</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.springframework.cloud</groupId>
-      <artifactId>spring-cloud-starter-bootstrap</artifactId>
-    </dependency>
     <dependency>
       <groupId>org.springframework.security</groupId>
       <artifactId>spring-security-config</artifactId>
@@ -190,6 +178,11 @@
       <artifactId>log4j-core</artifactId>
       <scope>runtime</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.logging.log4j</groupId>
+      <artifactId>log4j-slf4j-impl</artifactId>
+      <scope>runtime</scope>
+    </dependency>
     <dependency>
       <groupId>org.javassist</groupId>
       <artifactId>javassist</artifactId>
@@ -205,11 +198,6 @@
       <artifactId>jsoup</artifactId>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>org.springframework</groupId>
-      <artifactId>spring-test</artifactId>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-test-autoconfigure</artifactId>
@@ -218,10 +206,6 @@
   </dependencies>
   <build>
     <plugins>
-      <plugin>
-        <groupId>org.springframework.boot</groupId>
-        <artifactId>spring-boot-maven-plugin</artifactId>
-      </plugin>
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-assembly-plugin</artifactId>
@@ -281,20 +265,6 @@
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-failsafe-plugin</artifactId>
       </plugin>
-      <plugin>
-        <groupId>org.apache.maven.plugins</groupId>
-        <artifactId>maven-jar-plugin</artifactId>
-        <version>3.1.2</version>
-        <configuration>
-          <archive>
-            <manifest>
-              <!--suppress UnresolvedMavenProperty, MavenModelInspection -->
-              <mainClass>${start-class}</mainClass>
-              <addDefaultImplementationEntries>true</addDefaultImplementationEntries>
-            </manifest>
-          </archive>
-        </configuration>
-      </plugin>
       <plugin>
         <groupId>org.codehaus.mojo</groupId>
         <artifactId>build-helper-maven-plugin</artifactId>
@@ -322,17 +292,4 @@
       </extension>
     </extensions>
   </build>
-  <profiles>
-    <profile>
-      <id>docker</id>
-      <build>
-        <plugins>
-          <plugin>
-            <groupId>com.spotify</groupId>
-            <artifactId>docker-maven-plugin</artifactId>
-          </plugin>
-        </plugins>
-      </build>
-    </profile>
-  </profiles>
 </project>

balancer/src/main/dist/assembly.xml

@@ -1,4 +1,4 @@
-<assembly xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.3"
+<assembly xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.3" 
   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
   xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.3 http://maven.apache.org/xsd/assembly-1.1.3.xsd">
   <id>dist</id>
@@ -16,20 +16,19 @@
       </includes>
     </fileSet>
     <fileSet>
-      <directory>${project.build.directory}/lib</directory>
+      <directory>${project.build.directory}</directory>
       <outputDirectory>lib</outputDirectory>
       <includes>
-        <include>netty-tcnative-boringssl-static*.jar</include>
-        <include>netty-all*.jar</include>
+        <include>timely*.jar</include>
       </includes>
     </fileSet>
     <fileSet>
-      <directory>${project.build.directory}</directory>
-      <outputDirectory>bin</outputDirectory>
+      <directory>${project.build.directory}/lib</directory>
+      <outputDirectory>lib</outputDirectory>
       <includes>
-        <include>timely-balancer-*-exec.jar</include>
+        <include>*.jar</include>
       </includes>
-    </fileSet>
+     </fileSet>
     <fileSet>
       <directory>${project.basedir}/bin</directory>
       <outputDirectory>bin</outputDirectory>

balancer/src/main/docker/Dockerfile

@@ -1,11 +0,0 @@
-FROM azul/zulu-openjdk-alpine:11
-
-LABEL version=${project.version} \
-      run="docker run ${docker.image.prefix}${project.artifactId}:latest" \
-      description="${project.description}"
-
-ADD ${project.build.finalName}-exec.jar /app.jar
-RUN apk add libc6-compat curl
-
-EXPOSE 54322 54323 54324 54325
-ENTRYPOINT ["java","-jar","app.jar"]

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

@@ -0,0 +1,599 @@
+package timely.balancer;
+
+import static timely.Server.SERVICE_DISCOVERY_PATH;
+import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS;
+import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS_LOCK_PATH;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import javax.net.ssl.SSLContext;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.WriteBufferWaterMark;
+import io.netty.channel.epoll.EpollChannelOption;
+import io.netty.channel.epoll.EpollDatagramChannel;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollServerSocketChannel;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.DatagramChannel;
+import io.netty.channel.socket.ServerSocketChannel;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioDatagramChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.codec.DelimiterBasedFrameDecoder;
+import io.netty.handler.codec.Delimiters;
+import io.netty.handler.codec.http.HttpContentCompressor;
+import io.netty.handler.codec.http.HttpContentDecompressor;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.HttpRequestDecoder;
+import io.netty.handler.codec.http.HttpResponseEncoder;
+import io.netty.handler.codec.http.HttpServerCodec;
+import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
+import io.netty.handler.logging.LoggingHandler;
+import io.netty.handler.ssl.ClientAuth;
+import io.netty.handler.ssl.OpenSslServerContext;
+import io.netty.handler.ssl.OpenSslServerSessionContext;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslProvider;
+import io.netty.handler.ssl.util.SelfSignedCertificate;
+import io.netty.handler.stream.ChunkedWriteHandler;
+import io.netty.handler.timeout.IdleStateHandler;
+import io.netty.util.NettyRuntime;
+import io.netty.util.internal.SystemPropertyUtil;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.curator.RetryPolicy;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryForever;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.context.ConfigurableApplicationContext;
+import timely.auth.AuthCache;
+import timely.balancer.configuration.BalancerConfiguration;
+import timely.balancer.configuration.ClientSsl;
+import timely.balancer.configuration.SpringBootstrap;
+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.TcpHealthChecker;
+import timely.balancer.netty.http.HttpRelayHandler;
+import timely.balancer.netty.tcp.TcpRelayHandler;
+import timely.balancer.netty.udp.UdpRelayHandler;
+import timely.balancer.netty.ws.WsRelayHandler;
+import timely.balancer.resolver.BalancedMetricResolver;
+import timely.balancer.resolver.MetricResolver;
+import timely.client.http.HttpClient;
+import timely.configuration.ServerSsl;
+import timely.netty.http.HttpStaticFileServerHandler;
+import timely.netty.http.NonSslRedirectHandler;
+import timely.netty.http.TimelyExceptionHandler;
+import timely.netty.http.auth.X509LoginRequestHandler;
+import timely.netty.tcp.MetricsBufferDecoder;
+import timely.netty.tcp.TcpDecoder;
+import timely.netty.tcp.TcpVersionHandler;
+import timely.netty.udp.UdpDecoder;
+import timely.netty.udp.UdpPacketToByteBuf;
+import timely.netty.websocket.WebSocketFullRequestHandler;
+import timely.netty.websocket.WebSocketRequestDecoder;
+import timely.netty.websocket.subscription.WSTimelyExceptionHandler;
+
+public class Balancer {
+
+    final public static String LEADER_LATCH_PATH = "/timely/balancer/leader";
+    final public static String ASSIGNMENTS_LOCK_PATH = "/timely/balancer/assignments/lock";
+    final public static String ASSIGNMENTS_LAST_UPDATED_PATH = "/timely/balancer/assignments/lastUpdated";
+
+    private static final Logger LOG = LoggerFactory.getLogger(Balancer.class);
+    private static final int EPOLL_MIN_MAJOR_VERSION = 2;
+    private static final int EPOLL_MIN_MINOR_VERSION = 6;
+    private static final int EPOLL_MIN_PATCH_VERSION = 32;
+    private static final String OS_NAME = "os.name";
+    private static final String OS_VERSION = "os.version";
+    private static final String WS_PATH = "/websocket";
+    protected static final CountDownLatch LATCH = new CountDownLatch(1);
+    static ConfigurableApplicationContext applicationContext;
+
+    private final BalancerConfiguration balancerConfig;
+    private EventLoopGroup tcpWorkerGroup = null;
+    private EventLoopGroup tcpBossGroup = null;
+    private EventLoopGroup httpWorkerGroup = null;
+    private EventLoopGroup httpBossGroup = null;
+    private EventLoopGroup wsWorkerGroup = null;
+    private EventLoopGroup wsBossGroup = null;
+    private EventLoopGroup udpBossGroup = null;
+    private EventLoopGroup udpWorkerGroup = null;
+    private MetricResolver metricResolver;
+    protected Channel tcpChannelHandle = null;
+    protected Channel httpChannelHandle = null;
+    protected Channel wsChannelHandle = null;
+    protected List<Channel> udpChannelHandleList = new ArrayList<>();
+    private CuratorFramework curatorFramework;
+    protected volatile boolean shutdown = false;
+    private static final int DEFAULT_EVENT_LOOP_THREADS;
+
+    private String[] zkPaths = new String[] { LEADER_LATCH_PATH, ASSIGNMENTS_LAST_UPDATED_PATH, ASSIGNMENTS_LOCK_PATH,
+            SERVICE_DISCOVERY_PATH, NON_CACHED_METRICS, NON_CACHED_METRICS_LOCK_PATH };
+
+    static {
+        DEFAULT_EVENT_LOOP_THREADS = Math.max(1,
+                SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
+    }
+
+    public Balancer(BalancerConfiguration balancerConf) throws Exception {
+        this.balancerConfig = balancerConf;
+    }
+
+    protected static ConfigurableApplicationContext initializeContext(String[] args) {
+        SpringApplicationBuilder builder = new SpringApplicationBuilder(SpringBootstrap.class);
+        builder.web(WebApplicationType.NONE);
+        builder.registerShutdownHook(false);
+        return builder.run(args);
+    }
+
+    private void shutdownHook() {
+
+        final Runnable shutdownRunner = () -> {
+            if (!shutdown) {
+                shutdown();
+            }
+        };
+        final Thread hook = new Thread(shutdownRunner, "shutdown-hook-thread");
+        Runtime.getRuntime().addShutdownHook(hook);
+    }
+
+    public void shutdown() {
+        List<ChannelFuture> channelFutures = new ArrayList<>();
+
+        if (tcpChannelHandle != null) {
+            LOG.info("Closing tcpChannelHandle");
+            channelFutures.add(tcpChannelHandle.close());
+        }
+
+        if (httpChannelHandle != null) {
+            LOG.info("Closing httpChannelHandle");
+            channelFutures.add(httpChannelHandle.close());
+        }
+
+        if (wsChannelHandle != null) {
+            LOG.info("Closing wsChannelHandle");
+            channelFutures.add(wsChannelHandle.close());
+        }
+
+        int udpChannel = 1;
+        for (Channel c : udpChannelHandleList) {
+            LOG.info("Closing udpChannelHandle #" + udpChannel++);
+            channelFutures.add(c.close());
+        }
+
+        // wait for the channels to shutdown
+        channelFutures.forEach(f -> {
+            try {
+                f.get();
+            } catch (final Exception e) {
+                LOG.error("Channel:" + f.channel().config() + " -> " + e.getMessage(), e);
+            }
+        });
+
+        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));
+        }
+
+        if (tcpWorkerGroup != null) {
+            LOG.info("Shutting down tcpWorkerGroup");
+            groupFutures.add(tcpWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
+        }
+
+        if (httpBossGroup != null) {
+            LOG.info("Shutting down httpBossGroup");
+            groupFutures.add(httpBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
+        }
+
+        if (httpWorkerGroup != null) {
+            LOG.info("Shutting down httpWorkerGroup");
+            groupFutures.add(httpWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
+        }
+
+        if (wsBossGroup != null) {
+            LOG.info("Shutting down wsBossGroup");
+            groupFutures.add(wsBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
+        }
+
+        if (wsWorkerGroup != null) {
+            LOG.info("Shutting down wsWorkerGroup");
+            groupFutures.add(wsWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
+        }
+
+        if (udpBossGroup != null) {
+            LOG.info("Shutting down udpBossGroup");
+            groupFutures.add(udpBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
+        }
+
+        if (udpWorkerGroup != null) {
+            LOG.info("Shutting down udpWorkerGroup");
+            groupFutures.add(udpWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
+        }
+
+        groupFutures.parallelStream().forEach(f -> {
+            try {
+                f.get();
+            } catch (final Exception e) {
+                LOG.error("Group:" + f.toString() + " -> " + e.getMessage(), e);
+            }
+        });
+
+        try {
+            LOG.info("Closing WebSocketRequestDecoder");
+            WebSocketRequestDecoder.close();
+        } catch (Exception e) {
+            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();
+        }
+
+        this.shutdown = true;
+        LOG.info("Server shut down.");
+    }
+
+    private static boolean useEpoll() {
+
+        // Should we just return true if this is Linux and if we get an error
+        // during Epoll
+        // setup handle it there?
+        final String os = SystemPropertyUtil.get(OS_NAME).toLowerCase().trim();
+        final String[] version = SystemPropertyUtil.get(OS_VERSION).toLowerCase().trim().split("\\.");
+        if (os.startsWith("linux") && version.length >= 3) {
+            final int major = Integer.parseInt(version[0]);
+            if (major > EPOLL_MIN_MAJOR_VERSION) {
+                return true;
+            } else if (major == EPOLL_MIN_MAJOR_VERSION) {
+                final int minor = Integer.parseInt(version[1]);
+                if (minor > EPOLL_MIN_MINOR_VERSION) {
+                    return true;
+                } else if (minor == EPOLL_MIN_MINOR_VERSION) {
+                    final int patch = Integer.parseInt(version[2].substring(0, 2));
+                    return patch >= EPOLL_MIN_PATCH_VERSION;
+                } else {
+                    return false;
+                }
+            } else {
+                return false;
+            }
+        } else {
+            return false;
+        }
+    }
+
+    protected SslContext createSSLContext(BalancerConfiguration config) throws Exception {
+
+        ServerSsl sslCfg = config.getSecurity().getServerSsl();
+        Boolean generate = sslCfg.isUseGeneratedKeypair();
+        SslContextBuilder ssl;
+        if (generate) {
+            LOG.warn("Using generated self signed server certificate");
+            Date begin = new Date();
+            Date end = new Date(begin.getTime() + 86400000);
+            SelfSignedCertificate ssc = new SelfSignedCertificate("localhost", begin, end);
+            ssl = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey());
+        } else {
+            String cert = sslCfg.getCertificateFile();
+            String key = sslCfg.getKeyFile();
+            String keyPass = sslCfg.getKeyPassword();
+            if (null == cert || null == key) {
+                throw new IllegalArgumentException("Check your SSL properties, something is wrong.");
+            }
+            ssl = SslContextBuilder.forServer(new File(cert), new File(key), keyPass);
+        }
+
+        ssl.ciphers(sslCfg.getUseCiphers());
+
+        // Can't set to REQUIRE because the CORS pre-flight requests will fail.
+        ssl.clientAuth(ClientAuth.OPTIONAL);
+
+        Boolean useOpenSSL = sslCfg.isUseOpenssl();
+        if (useOpenSSL) {
+            ssl.sslProvider(SslProvider.OPENSSL);
+        } else {
+            ssl.sslProvider(SslProvider.JDK);
+        }
+        String trustStore = sslCfg.getTrustStoreFile();
+        if (null != trustStore) {
+            if (!trustStore.isEmpty()) {
+                ssl.trustManager(new File(trustStore));
+            }
+        }
+        return ssl.build();
+    }
+
+    protected SSLContext createSSLClientContext(BalancerConfiguration config) throws Exception {
+
+        ClientSsl ssl = config.getSecurity().getClientSsl();
+        return HttpClient.getSSLContext(ssl.getTrustStoreFile(), ssl.getTrustStoreType(), ssl.getTrustStorePassword(),
+                ssl.getKeyFile(), ssl.getKeyType(), ssl.getKeyPassword());
+    }
+
+    private void ensureZkPaths(CuratorFramework curatorFramework, String[] paths) {
+        for (String s : paths) {
+            try {
+                Stat stat = curatorFramework.checkExists().forPath(s);
+                if (stat == null) {
+                    curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT)
+                            .forPath(s);
+                }
+            } catch (Exception e) {
+                LOG.info(e.getMessage());
+            }
+
+        }
+    }
+
+    public void run() throws Exception {
+
+        // start curator framework
+        RetryPolicy retryPolicy = new RetryForever(1000);
+        int timeout = Long.valueOf(AccumuloConfiguration.getTimeInMillis(balancerConfig.getZooKeeper().getTimeout()))
+                .intValue();
+        curatorFramework = CuratorFrameworkFactory.newClient(balancerConfig.getZooKeeper().getServers(), timeout, 10000,
+                retryPolicy);
+        curatorFramework.start();
+        ensureZkPaths(curatorFramework, zkPaths);
+
+        AuthCache.configure(balancerConfig.getSecurity());
+        final boolean useEpoll = useEpoll();
+        Class<? extends ServerSocketChannel> channelClass;
+        Class<? extends Channel> datagramChannelClass;
+        if (useEpoll) {
+            tcpWorkerGroup = new EpollEventLoopGroup();
+            tcpBossGroup = new EpollEventLoopGroup();
+            httpWorkerGroup = new EpollEventLoopGroup();
+            httpBossGroup = new EpollEventLoopGroup();
+            wsWorkerGroup = new EpollEventLoopGroup();
+            wsBossGroup = new EpollEventLoopGroup();
+            udpWorkerGroup = new EpollEventLoopGroup();
+            udpBossGroup = new EpollEventLoopGroup();
+            channelClass = EpollServerSocketChannel.class;
+            datagramChannelClass = EpollDatagramChannel.class;
+        } else {
+            tcpWorkerGroup = new NioEventLoopGroup();
+            tcpBossGroup = new NioEventLoopGroup();
+            httpWorkerGroup = new NioEventLoopGroup();
+            httpBossGroup = new NioEventLoopGroup();
+            wsWorkerGroup = new NioEventLoopGroup();
+            wsBossGroup = new NioEventLoopGroup();
+            udpWorkerGroup = new NioEventLoopGroup();
+            udpBossGroup = new NioEventLoopGroup();
+            channelClass = NioServerSocketChannel.class;
+            datagramChannelClass = NioDatagramChannel.class;
+        }
+        LOG.info("Using channel class {}", channelClass.getSimpleName());
+
+        List<TcpClientPool> tcpClientPools = new ArrayList<>();
+        for (int x = 0; x < balancerConfig.getServer().getNumTcpPools(); x++) {
+            tcpClientPools.add(new TcpClientPool(this.balancerConfig));
+        }
+        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);
+        tcpServer.channel(channelClass);
+        tcpServer.handler(new LoggingHandler());
+        tcpServer.childHandler(setupTcpChannel(metricResolver, tcpClientPools));
+        tcpServer.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
+        tcpServer.option(ChannelOption.SO_BACKLOG, 128);
+        tcpServer.option(ChannelOption.SO_KEEPALIVE, true);
+        final int tcpPort = balancerConfig.getServer().getTcpPort();
+        final String tcpIp = balancerConfig.getServer().getIp();
+        tcpChannelHandle = tcpServer.bind(tcpIp, tcpPort).sync().channel();
+        final String tcpAddress = ((InetSocketAddress) tcpChannelHandle.localAddress()).getAddress().getHostAddress();
+
+        final int httpPort = balancerConfig.getHttp().getPort();
+        final String httpIp = balancerConfig.getHttp().getIp();
+        SslContext sslCtx = createSSLContext(balancerConfig);
+        if (sslCtx instanceof OpenSslServerContext) {
+            OpenSslServerContext openssl = (OpenSslServerContext) sslCtx;
+            String application = "Timely_" + httpPort;
+            OpenSslServerSessionContext opensslCtx = openssl.sessionContext();
+            opensslCtx.setSessionCacheEnabled(true);
+            opensslCtx.setSessionCacheSize(128);
+            opensslCtx.setSessionIdContext(application.getBytes(StandardCharsets.UTF_8));
+            opensslCtx.setSessionTimeout(balancerConfig.getSecurity().getSessionMaxAge());
+        }
+        SSLContext clientSSLContext = createSSLClientContext(balancerConfig);
+        final ServerBootstrap httpServer = new ServerBootstrap();
+        httpServer.group(httpBossGroup, httpWorkerGroup);
+        httpServer.channel(channelClass);
+        httpServer.handler(new LoggingHandler());
+        HttpClientPool httpClientPool = new HttpClientPool(balancerConfig.getSecurity(), balancerConfig.getHttp(),
+                clientSSLContext);
+        httpServer.childHandler(setupHttpChannel(balancerConfig, sslCtx, metricResolver, httpClientPool));
+        httpServer.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
+        httpServer.option(ChannelOption.SO_BACKLOG, 128);
+        httpServer.option(ChannelOption.SO_KEEPALIVE, true);
+        httpChannelHandle = httpServer.bind(httpIp, httpPort).sync().channel();
+        final String httpAddress = ((InetSocketAddress) httpChannelHandle.localAddress()).getAddress().getHostAddress();
+
+        final int wsPort = balancerConfig.getWebsocket().getPort();
+        final String wsIp = balancerConfig.getWebsocket().getIp();
+        final ServerBootstrap wsServer = new ServerBootstrap();
+        wsServer.group(wsBossGroup, wsWorkerGroup);
+        wsServer.channel(channelClass);
+        wsServer.handler(new LoggingHandler());
+        WsClientPool wsClientPool = new WsClientPool(balancerConfig, clientSSLContext);
+        wsServer.childHandler(setupWSChannel(balancerConfig, sslCtx, metricResolver, wsClientPool));
+        wsServer.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
+        wsServer.option(ChannelOption.SO_BACKLOG, 128);
+        wsServer.option(ChannelOption.SO_KEEPALIVE, true);
+        /* Not sure if next two lines are necessary */
+        wsServer.option(ChannelOption.SO_SNDBUF, 1048576);
+        wsServer.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(620145, 838860));
+        wsChannelHandle = wsServer.bind(wsIp, wsPort).sync().channel();
+        final String wsAddress = ((InetSocketAddress) wsChannelHandle.localAddress()).getAddress().getHostAddress();
+
+        final int udpPort = balancerConfig.getServer().getUdpPort();
+        final String udpIp = balancerConfig.getServer().getIp();
+        UdpClientPool udpClientPool = new UdpClientPool(balancerConfig);
+        for (int n = 1; n < DEFAULT_EVENT_LOOP_THREADS; n++) {
+            final Bootstrap udpServer = new Bootstrap();
+            udpServer.group(udpBossGroup);
+            udpServer.channel(datagramChannelClass);
+            udpServer.handler(setupUdpChannel(metricResolver, udpClientPool));
+            udpServer.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
+            udpServer.option(EpollChannelOption.SO_REUSEADDR, true);
+            udpServer.option(EpollChannelOption.SO_REUSEPORT, true);
+            udpChannelHandleList.add(udpServer.bind(udpIp, udpPort).sync().channel());
+        }
+        shutdownHook();
+        LOG.info(
+                "Server started. Listening on {}:{} for TCP traffic, {}:{} for HTTP traffic, {}:{} for WebSocket traffic, and {}:{} for UDP traffic",
+                tcpAddress, tcpPort, httpAddress, httpPort, wsAddress, wsPort, wsAddress, udpPort);
+    }
+
+    public static void main(String[] args) throws Exception {
+
+        Balancer.applicationContext = Balancer.initializeContext(args);
+        BalancerConfiguration balancerConf = Balancer.applicationContext.getBean(BalancerConfiguration.class);
+
+        Balancer balancer = new Balancer(balancerConf);
+        try {
+            balancer.run();
+            LATCH.await();
+        } catch (final InterruptedException e) {
+            LOG.info("Server shutting down.");
+        } catch (Exception e) {
+            LOG.error("Error running server.", e);
+        } finally {
+            try {
+                balancer.shutdown();
+            } catch (Exception e) {
+                System.exit(1);
+            }
+        }
+    }
+
+    protected ChannelHandler setupHttpChannel(BalancerConfiguration balancerConfig, SslContext sslCtx,
+            MetricResolver metricResolver, HttpClientPool httpClientPool) {
+
+        return new ChannelInitializer<SocketChannel>() {
+
+            @Override
+            protected void initChannel(SocketChannel ch) throws Exception {
+
+                ch.pipeline().addLast("ssl", new NonSslRedirectHandler(balancerConfig.getHttp(), sslCtx));
+                ch.pipeline().addLast("encoder", new HttpResponseEncoder());
+                ch.pipeline().addLast("decoder", new HttpRequestDecoder());
+                ch.pipeline().addLast("compressor", new HttpContentCompressor());
+                ch.pipeline().addLast("decompressor", new HttpContentDecompressor());
+                ch.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
+                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().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().setIgnoreSslHandshakeErrors(
+                        balancerConfig.getSecurity().getServerSsl().isUseGeneratedKeypair()));
+            }
+        };
+    }
+
+    protected ChannelHandler setupUdpChannel(MetricResolver metricResolver, UdpClientPool udpClientPool) {
+        return new ChannelInitializer<DatagramChannel>() {
+
+            @Override
+            protected void initChannel(DatagramChannel ch) throws Exception {
+                ch.pipeline().addLast("logger", new LoggingHandler());
+                ch.pipeline().addLast("packetDecoder", new UdpPacketToByteBuf());
+                ch.pipeline().addLast("buffer", new MetricsBufferDecoder());
+                ch.pipeline().addLast("frame", new DelimiterBasedFrameDecoder(65536, true, Delimiters.lineDelimiter()));
+                ch.pipeline().addLast("putDecoder", new UdpDecoder());
+                ch.pipeline().addLast("udpRelayHandler", new UdpRelayHandler(metricResolver, udpClientPool));
+            }
+        };
+    }
+
+    protected ChannelHandler setupTcpChannel(MetricResolver metricResolver, List<TcpClientPool> tcpClientPools) {
+        return new ChannelInitializer<SocketChannel>() {
+
+            @Override
+            protected void initChannel(SocketChannel ch) throws Exception {
+                ch.pipeline().addLast("buffer", new MetricsBufferDecoder());
+                ch.pipeline().addLast("frame", new DelimiterBasedFrameDecoder(65536, true, Delimiters.lineDelimiter()));
+                ch.pipeline().addLast("putDecoder", new TcpDecoder());
+                ch.pipeline().addLast("tcpRelayHandler", new TcpRelayHandler(metricResolver, tcpClientPools));
+                ch.pipeline().addLast("versionHandler", new TcpVersionHandler());
+            }
+        };
+    }
+
+    protected ChannelHandler setupWSChannel(BalancerConfiguration balancerConfig, SslContext sslCtx,
+            MetricResolver metricResolver, WsClientPool wsClientPool) {
+        return new ChannelInitializer<SocketChannel>() {
+
+            @Override
+            protected void initChannel(SocketChannel ch) throws Exception {
+                ch.pipeline().addLast("ssl", sslCtx.newHandler(ch.alloc()));
+                ch.pipeline().addLast("httpServer", new HttpServerCodec());
+                ch.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
+                ch.pipeline().addLast("sessionExtractor", new WebSocketFullRequestHandler());
+
+                ch.pipeline().addLast("idle-handler",
+                        new IdleStateHandler(balancerConfig.getWebsocket().getTimeout(), 0, 0));
+                ch.pipeline().addLast("ws-protocol",
+                        new WebSocketServerProtocolHandler(WS_PATH, null, true, 65536, false, true));
+                ch.pipeline().addLast("wsDecoder", new WebSocketRequestDecoder(balancerConfig.getSecurity()));
+                ch.pipeline().addLast("httpRelay", new WsRelayHandler(balancerConfig, metricResolver, wsClientPool));
+                ch.pipeline().addLast("error", new WSTimelyExceptionHandler());
+            }
+        };
+    }
+}

balancer/src/main/java/timely/balancer/TimelyBalancer.java

@@ -1,14 +0,0 @@
-package timely.balancer;
-
-import org.springframework.boot.WebApplicationType;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.boot.autoconfigure.web.servlet.error.ErrorMvcAutoConfiguration;
-import org.springframework.boot.builder.SpringApplicationBuilder;
-
-@SpringBootApplication(scanBasePackages = {"timely.balancer", "timely.common"}, exclude = {ErrorMvcAutoConfiguration.class})
-public class TimelyBalancer {
-
-    public static void main(String[] args) {
-        new SpringApplicationBuilder(TimelyBalancer.class).web(WebApplicationType.NONE).run(args);
-    }
-}

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

@@ -1,526 +0,0 @@
-package timely.balancer.component;
-
-import static org.apache.accumulo.core.conf.ConfigurationTypeHelper.getTimeInMillis;
-import static timely.Constants.NON_CACHED_METRICS;
-import static timely.Constants.NON_CACHED_METRICS_LOCK_PATH;
-import static timely.Constants.SERVICE_DISCOVERY_PATH;
-
-import java.net.InetSocketAddress;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Date;
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
-import javax.net.ssl.SSLContext;
-
-import org.apache.curator.RetryPolicy;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.CuratorFrameworkFactory;
-import org.apache.curator.retry.RetryForever;
-import org.apache.hadoop.util.StringUtils;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.data.Stat;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.context.ApplicationContext;
-import org.springframework.stereotype.Component;
-
-import io.netty.bootstrap.AbstractBootstrap;
-import io.netty.bootstrap.Bootstrap;
-import io.netty.bootstrap.ServerBootstrap;
-import io.netty.buffer.PooledByteBufAllocator;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelHandler;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelOption;
-import io.netty.channel.EventLoopGroup;
-import io.netty.channel.epoll.EpollChannelOption;
-import io.netty.channel.epoll.EpollDatagramChannel;
-import io.netty.channel.epoll.EpollEventLoopGroup;
-import io.netty.channel.epoll.EpollServerSocketChannel;
-import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.channel.socket.DatagramChannel;
-import io.netty.channel.socket.ServerSocketChannel;
-import io.netty.channel.socket.SocketChannel;
-import io.netty.channel.socket.nio.NioDatagramChannel;
-import io.netty.channel.socket.nio.NioServerSocketChannel;
-import io.netty.handler.codec.DelimiterBasedFrameDecoder;
-import io.netty.handler.codec.Delimiters;
-import io.netty.handler.codec.http.HttpContentCompressor;
-import io.netty.handler.codec.http.HttpContentDecompressor;
-import io.netty.handler.codec.http.HttpObjectAggregator;
-import io.netty.handler.codec.http.HttpRequestDecoder;
-import io.netty.handler.codec.http.HttpResponseEncoder;
-import io.netty.handler.codec.http.HttpServerCodec;
-import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
-import io.netty.handler.logging.LoggingHandler;
-import io.netty.handler.ssl.ClientAuth;
-import io.netty.handler.ssl.OpenSslServerContext;
-import io.netty.handler.ssl.OpenSslServerSessionContext;
-import io.netty.handler.ssl.SslContext;
-import io.netty.handler.ssl.SslContextBuilder;
-import io.netty.handler.ssl.util.SelfSignedCertificate;
-import io.netty.handler.stream.ChunkedWriteHandler;
-import io.netty.handler.timeout.IdleStateHandler;
-import io.netty.util.NettyRuntime;
-import io.netty.util.internal.SystemPropertyUtil;
-import timely.auth.JWTTokenHandler;
-import timely.balancer.configuration.BalancerHttpProperties;
-import timely.balancer.configuration.BalancerProperties;
-import timely.balancer.configuration.BalancerServerProperties;
-import timely.balancer.configuration.BalancerWebsocketProperties;
-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.netty.http.HttpRelayHandler;
-import timely.balancer.netty.tcp.TcpRelayHandler;
-import timely.balancer.netty.udp.UdpRelayHandler;
-import timely.balancer.netty.ws.WsRelayHandler;
-import timely.client.http.HttpClient;
-import timely.common.component.AuthenticationService;
-import timely.common.configuration.SecurityProperties;
-import timely.common.configuration.SslClientProperties;
-import timely.common.configuration.SslHelper;
-import timely.common.configuration.SslServerProperties;
-import timely.common.configuration.ZookeeperProperties;
-import timely.netty.http.HttpStaticFileServerHandler;
-import timely.netty.http.NonSslRedirectHandler;
-import timely.netty.http.TimelyExceptionHandler;
-import timely.netty.http.auth.X509LoginRequestHandler;
-import timely.netty.tcp.MetricsBufferDecoder;
-import timely.netty.tcp.TcpDecoder;
-import timely.netty.tcp.TcpVersionHandler;
-import timely.netty.udp.UdpDecoder;
-import timely.netty.udp.UdpPacketToByteBuf;
-import timely.netty.websocket.WebSocketFullRequestHandler;
-import timely.netty.websocket.WebSocketRequestDecoder;
-import timely.netty.websocket.subscription.WSTimelyExceptionHandler;
-
-@Component
-@ConditionalOnMissingBean(Balancer.class)
-public class Balancer {
-
-    final public static String LEADER_LATCH_PATH = "/timely/balancer/leader";
-    final public static String ASSIGNMENTS_LOCK_PATH = "/timely/balancer/assignments/lock";
-    final public static String ASSIGNMENTS_LAST_UPDATED_PATH = "/timely/balancer/assignments/lastUpdated";
-
-    private static final Logger log = LoggerFactory.getLogger(Balancer.class);
-    private static final String EPOLL_MIN_VERSION = "2.7";
-    private static final String OS_NAME = "os.name";
-    private static final String OS_VERSION = "os.version";
-    private static final String WS_PATH = "/websocket";
-    protected static final CountDownLatch LATCH = new CountDownLatch(1);
-
-    private BalancerProperties balancerProperties;
-    private SecurityProperties securityProperties;
-    private SslServerProperties sslServerProperties;
-    private SslClientProperties sslClientProperties;
-    private ZookeeperProperties zookeeperProperties;
-    private BalancerServerProperties balancerServerProperties;
-    private BalancerHttpProperties balancerHttpProperties;
-    private BalancerWebsocketProperties balancerWebsocketProperties;
-    private ApplicationContext applicationContext;
-    private AuthenticationService authenticationService;
-    private CuratorFramework curatorFramework;
-    private int quietPeriod;
-
-    private EventLoopGroup tcpWorkerGroup = null;
-    private EventLoopGroup tcpBossGroup = null;
-    private EventLoopGroup httpWorkerGroup = null;
-    private EventLoopGroup httpBossGroup = null;
-    private EventLoopGroup wsWorkerGroup = null;
-    private EventLoopGroup wsBossGroup = null;
-    private EventLoopGroup udpBossGroup = null;
-    private EventLoopGroup udpWorkerGroup = null;
-    private MetricResolver metricResolver;
-    protected Channel tcpChannelHandle = null;
-    protected Channel httpChannelHandle = null;
-    protected Channel wsChannelHandle = null;
-    protected List<Channel> udpChannelHandleList = new ArrayList<>();
-    protected int DEFAULT_EVENT_LOOP_THREADS;
-
-    private String[] zkPaths = new String[] {LEADER_LATCH_PATH, ASSIGNMENTS_LAST_UPDATED_PATH, ASSIGNMENTS_LOCK_PATH, SERVICE_DISCOVERY_PATH,
-            NON_CACHED_METRICS, NON_CACHED_METRICS_LOCK_PATH};
-
-    private static boolean useEpoll() {
-        final String osName = SystemPropertyUtil.get(OS_NAME).toLowerCase().trim();
-        final String osVersion = SystemPropertyUtil.get(OS_VERSION).toLowerCase();
-        // split at periods, keep the first two, join with a period
-        final String majMinVers = StringUtils.join(".", Arrays.stream(osVersion.split("\\.")).limit(2).collect(Collectors.toList()));
-        if (osName.startsWith("linux")) {
-            Runtime.Version currentVersion = Runtime.Version.parse(majMinVers);
-            Runtime.Version epollMinVersion = Runtime.Version.parse(EPOLL_MIN_VERSION);
-            return currentVersion.compareTo(epollMinVersion) > 0;
-        } else {
-            return false;
-        }
-    }
-
-    public Balancer(ApplicationContext applicationContext, AuthenticationService authenticationService, MetricResolver metricResolver,
-                    BalancerProperties balancerProperties, ZookeeperProperties zookeeperProperties, SecurityProperties securityProperties,
-                    SslServerProperties sslServerProperties, SslClientProperties sslClientProperties, BalancerServerProperties balancerServerProperties,
-                    BalancerHttpProperties balancerHttpProperties, BalancerWebsocketProperties balancerWebsocketProperties) {
-        this.quietPeriod = balancerServerProperties.getShutdownQuietPeriod();
-        this.applicationContext = applicationContext;
-        this.authenticationService = authenticationService;
-        this.metricResolver = metricResolver;
-        this.balancerProperties = balancerProperties;
-        this.zookeeperProperties = zookeeperProperties;
-        this.securityProperties = securityProperties;
-        this.sslServerProperties = sslServerProperties;
-        this.sslClientProperties = sslClientProperties;
-        this.balancerServerProperties = balancerServerProperties;
-        this.balancerHttpProperties = balancerHttpProperties;
-        this.balancerWebsocketProperties = balancerWebsocketProperties;
-        DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
-    }
-
-    @PostConstruct
-    public void startup() {
-        log.info("Starting {}", this.getClass().getSimpleName());
-        try {
-            // start curator framework
-            RetryPolicy retryPolicy = new RetryForever(1000);
-            int timeout = Long.valueOf(getTimeInMillis(zookeeperProperties.getTimeout())).intValue();
-            curatorFramework = CuratorFrameworkFactory.newClient(zookeeperProperties.getServers(), timeout, 10000, retryPolicy);
-            curatorFramework.start();
-            ensureZkPaths(curatorFramework, zkPaths);
-
-            JWTTokenHandler.init(securityProperties, null);
-            final boolean useEpoll = useEpoll();
-            Class<? extends ServerSocketChannel> channelClass;
-            Class<? extends Channel> datagramChannelClass;
-            if (useEpoll) {
-                tcpWorkerGroup = new EpollEventLoopGroup();
-                tcpBossGroup = new EpollEventLoopGroup();
-                httpWorkerGroup = new EpollEventLoopGroup();
-                httpBossGroup = new EpollEventLoopGroup();
-                wsWorkerGroup = new EpollEventLoopGroup();
-                wsBossGroup = new EpollEventLoopGroup();
-                udpWorkerGroup = new EpollEventLoopGroup();
-                udpBossGroup = new EpollEventLoopGroup();
-                channelClass = EpollServerSocketChannel.class;
-                datagramChannelClass = EpollDatagramChannel.class;
-            } else {
-                tcpWorkerGroup = new NioEventLoopGroup();
-                tcpBossGroup = new NioEventLoopGroup();
-                httpWorkerGroup = new NioEventLoopGroup();
-                httpBossGroup = new NioEventLoopGroup();
-                wsWorkerGroup = new NioEventLoopGroup();
-                wsBossGroup = new NioEventLoopGroup();
-                udpWorkerGroup = new NioEventLoopGroup();
-                udpBossGroup = new NioEventLoopGroup();
-                channelClass = NioServerSocketChannel.class;
-                datagramChannelClass = NioDatagramChannel.class;
-            }
-            log.info("Using channel class {}", channelClass.getSimpleName());
-
-            List<TcpClientPool> tcpClientPools = new ArrayList<>();
-            for (int x = 0; x < balancerServerProperties.getNumTcpPools(); x++) {
-                tcpClientPools.add(new TcpClientPool(balancerServerProperties));
-            }
-
-            final ServerBootstrap tcpServer = new ServerBootstrap();
-            tcpServer.group(tcpBossGroup, tcpWorkerGroup);
-            tcpServer.channel(channelClass);
-            tcpServer.handler(new LoggingHandler());
-            tcpServer.childHandler(setupTcpChannel(metricResolver, tcpClientPools));
-            tcpServer.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
-            tcpServer.option(ChannelOption.SO_BACKLOG, 128);
-            tcpServer.childOption(ChannelOption.SO_KEEPALIVE, true);
-            final int tcpPort = balancerServerProperties.getTcpPort();
-            final String tcpIp = balancerServerProperties.getIp();
-            tcpChannelHandle = bind(tcpServer, tcpIp, tcpPort);
-            final String tcpAddress = ((InetSocketAddress) tcpChannelHandle.localAddress()).getAddress().getHostAddress();
-
-            final int httpPort = balancerHttpProperties.getPort();
-            final String httpIp = balancerHttpProperties.getIp();
-            SslContext sslCtx = createSSLContext(sslServerProperties);
-            if (sslCtx instanceof OpenSslServerContext) {
-                OpenSslServerContext openssl = (OpenSslServerContext) sslCtx;
-                String application = "Timely_" + httpPort;
-                OpenSslServerSessionContext opensslCtx = openssl.sessionContext();
-                opensslCtx.setSessionCacheEnabled(true);
-                opensslCtx.setSessionCacheSize(128);
-                opensslCtx.setSessionIdContext(application.getBytes(StandardCharsets.UTF_8));
-                opensslCtx.setSessionTimeout(securityProperties.getSessionMaxAge());
-            }
-            SSLContext clientSSLContext = createSSLClientContext(sslClientProperties);
-            final ServerBootstrap httpServer = new ServerBootstrap();
-            httpServer.group(httpBossGroup, httpWorkerGroup);
-            httpServer.channel(channelClass);
-            httpServer.handler(new LoggingHandler());
-            HttpClientPool httpClientPool = new HttpClientPool(securityProperties, sslClientProperties, balancerHttpProperties, clientSSLContext);
-            httpServer.childHandler(setupHttpChannel(balancerHttpProperties, sslServerProperties, sslCtx, metricResolver, httpClientPool));
-            httpServer.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
-            httpServer.option(ChannelOption.SO_BACKLOG, 128);
-            httpServer.childOption(ChannelOption.SO_KEEPALIVE, true);
-            httpChannelHandle = bind(httpServer, httpIp, httpPort);
-            final String httpAddress = ((InetSocketAddress) httpChannelHandle.localAddress()).getAddress().getHostAddress();
-
-            final int wsPort = balancerWebsocketProperties.getPort();
-            final String wsIp = balancerWebsocketProperties.getIp();
-            final ServerBootstrap wsServer = new ServerBootstrap();
-            wsServer.group(wsBossGroup, wsWorkerGroup);
-            wsServer.channel(channelClass);
-            wsServer.handler(new LoggingHandler());
-            WsClientPool wsClientPool = new WsClientPool(balancerProperties, balancerWebsocketProperties, sslClientProperties, clientSSLContext);
-            wsServer.childHandler(setupWSChannel(balancerWebsocketProperties, securityProperties, sslCtx, metricResolver, wsClientPool));
-            wsServer.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
-            wsServer.option(ChannelOption.SO_BACKLOG, 128);
-            wsServer.childOption(ChannelOption.SO_KEEPALIVE, true);
-            wsChannelHandle = bind(wsServer, wsIp, wsPort);
-            final String wsAddress = ((InetSocketAddress) wsChannelHandle.localAddress()).getAddress().getHostAddress();
-
-            final int udpPort = balancerServerProperties.getUdpPort();
-            final String udpIp = balancerServerProperties.getIp();
-            UdpClientPool udpClientPool = new UdpClientPool(balancerServerProperties);
-            for (int n = 0; n < DEFAULT_EVENT_LOOP_THREADS; n++) {
-                final Bootstrap udpServer = new Bootstrap();
-                udpServer.group(udpBossGroup);
-                udpServer.channel(datagramChannelClass);
-                udpServer.handler(setupUdpChannel(metricResolver, udpClientPool));
-                udpServer.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
-                udpServer.option(EpollChannelOption.SO_REUSEADDR, true);
-                udpServer.option(EpollChannelOption.SO_REUSEPORT, true);
-                udpChannelHandleList.add(bind(udpServer, udpIp, udpPort));
-            }
-            log.info("Balancer started. Listening on {}:{} for TCP traffic, {}:{} for HTTP traffic, {}:{} for WebSocket traffic, and {}:{} for UDP traffic",
-                            tcpAddress, tcpPort, httpAddress, httpPort, wsAddress, wsPort, wsAddress, udpPort);
-        } catch (Exception e) {
-            log.error(e.getMessage(), e);
-            SpringApplication.exit(applicationContext, () -> 0);
-        }
-    }
-
-    protected Channel bind(AbstractBootstrap server, String ip, int port) {
-        Channel channel = null;
-        try {
-            channel = server.bind(ip, port).sync().channel();
-        } catch (InterruptedException e) {
-            log.error("Failed to bind to {}:{} -- {}", ip, port, e.getMessage());
-        }
-        if (channel == null) {
-            throw new IllegalStateException("Failed to bind to port:" + ip + ":" + port);
-        }
-        return channel;
-    }
-
-    @PreDestroy
-    public void shutdown() {
-        List<ChannelFuture> channelFutures = new ArrayList<>();
-
-        if (tcpChannelHandle != null) {
-            log.info("Closing tcpChannelHandle");
-            channelFutures.add(tcpChannelHandle.close());
-        }
-
-        if (httpChannelHandle != null) {
-            log.info("Closing httpChannelHandle");
-            channelFutures.add(httpChannelHandle.close());
-        }
-
-        if (wsChannelHandle != null) {
-            log.info("Closing wsChannelHandle");
-            channelFutures.add(wsChannelHandle.close());
-        }
-
-        int udpChannel = 1;
-        for (Channel c : udpChannelHandleList) {
-            log.info("Closing udpChannelHandle #" + udpChannel++);
-            channelFutures.add(c.close());
-        }
-
-        // wait for the channels to shutdown
-        channelFutures.forEach(f -> {
-            try {
-                f.get();
-            } catch (final Exception e) {
-                log.error("Channel:" + f.channel().config() + " -> " + e.getMessage(), e);
-            }
-        });
-
-        List<Future<?>> groupFutures = new ArrayList<>();
-
-        if (tcpBossGroup != null) {
-            log.info("Shutting down tcpBossGroup");
-            groupFutures.add(tcpBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
-        }
-
-        if (tcpWorkerGroup != null) {
-            log.info("Shutting down tcpWorkerGroup");
-            groupFutures.add(tcpWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
-        }
-
-        if (httpBossGroup != null) {
-            log.info("Shutting down httpBossGroup");
-            groupFutures.add(httpBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
-        }
-
-        if (httpWorkerGroup != null) {
-            log.info("Shutting down httpWorkerGroup");
-            groupFutures.add(httpWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
-        }
-
-        if (wsBossGroup != null) {
-            log.info("Shutting down wsBossGroup");
-            groupFutures.add(wsBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
-        }
-
-        if (wsWorkerGroup != null) {
-            log.info("Shutting down wsWorkerGroup");
-            groupFutures.add(wsWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
-        }
-
-        if (udpBossGroup != null) {
-            log.info("Shutting down udpBossGroup");
-            groupFutures.add(udpBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
-        }
-
-        if (udpWorkerGroup != null) {
-            log.info("Shutting down udpWorkerGroup");
-            groupFutures.add(udpWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
-        }
-
-        groupFutures.parallelStream().forEach(f -> {
-            try {
-                f.get();
-            } catch (final Exception e) {
-                log.error("Group:" + f.toString() + " -> " + e.getMessage(), e);
-            }
-        });
-
-        try {
-            log.info("Closing WebSocketRequestDecoder");
-            WebSocketRequestDecoder.close();
-        } catch (Exception e) {
-            log.error("Error closing WebSocketRequestDecoder during shutdown", e);
-        }
-        log.info("{} shut down.", this.getClass().getSimpleName());
-    }
-
-    protected SslContext createSSLContext(SslServerProperties sslServerProperties) throws Exception {
-
-        Boolean generate = sslServerProperties.isUseGeneratedKeypair();
-        SslContextBuilder ssl;
-        if (generate) {
-            log.warn("Using generated self signed server certificate");
-            Date begin = new Date();
-            Date end = new Date(begin.getTime() + 86400000);
-            SelfSignedCertificate ssc = new SelfSignedCertificate("localhost", begin, end);
-            ssl = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey());
-        } else {
-            ssl = SslHelper.getSslContextBuilder(sslServerProperties);
-        }
-        // Can't set to REQUIRE because the CORS pre-flight requests will fail.
-        ssl.clientAuth(ClientAuth.OPTIONAL);
-        return ssl.build();
-    }
-
-    protected SSLContext createSSLClientContext(SslClientProperties sslClientProperties) {
-
-        SslClientProperties ssl = sslClientProperties;
-        return HttpClient.getSSLContext(ssl.getTrustStoreFile(), ssl.getTrustStoreType(), ssl.getTrustStorePassword(), ssl.getKeyStoreFile(),
-                        ssl.getKeyStoreType(), ssl.getKeyStorePassword());
-    }
-
-    private void ensureZkPaths(CuratorFramework curatorFramework, String[] paths) {
-        for (String s : paths) {
-            try {
-                Stat stat = curatorFramework.checkExists().forPath(s);
-                if (stat == null) {
-                    curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath(s);
-                }
-            } catch (Exception e) {
-                log.info(e.getMessage());
-            }
-
-        }
-    }
-
-    protected ChannelHandler setupHttpChannel(BalancerHttpProperties balancerHttpProperties, SslServerProperties sslServerProperties, SslContext sslCtx,
-                    MetricResolver metricResolver, HttpClientPool httpClientPool) {
-
-        return new ChannelInitializer<SocketChannel>() {
-
-            @Override
-            protected void initChannel(SocketChannel ch) {
-
-                ch.pipeline().addLast("ssl", new NonSslRedirectHandler(balancerHttpProperties, sslCtx));
-                ch.pipeline().addLast("encoder", new HttpResponseEncoder());
-                ch.pipeline().addLast("decoder", new HttpRequestDecoder(4096, 32768, 8192, true, 128));
-                ch.pipeline().addLast("compressor", new HttpContentCompressor());
-                ch.pipeline().addLast("decompressor", new HttpContentDecompressor());
-                ch.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
-                ch.pipeline().addLast("chunker", new ChunkedWriteHandler());
-                ch.pipeline().addLast("queryDecoder",
-                                new timely.netty.http.HttpRequestDecoder(authenticationService, securityProperties, balancerHttpProperties));
-                ch.pipeline().addLast("fileServer",
-                                new HttpStaticFileServerHandler().setIgnoreSslHandshakeErrors(sslServerProperties.isIgnoreSslHandshakeErrors()));
-                ch.pipeline().addLast("login", new X509LoginRequestHandler(authenticationService, securityProperties, balancerHttpProperties));
-                ch.pipeline().addLast("httpRelay", new HttpRelayHandler(metricResolver, httpClientPool));
-                ch.pipeline().addLast("error", new TimelyExceptionHandler().setIgnoreSslHandshakeErrors(sslServerProperties.isIgnoreSslHandshakeErrors()));
-            }
-        };
-    }
-
-    protected ChannelHandler setupUdpChannel(MetricResolver metricResolver, UdpClientPool udpClientPool) {
-        return new ChannelInitializer<DatagramChannel>() {
-
-            @Override
-            protected void initChannel(DatagramChannel ch) throws Exception {
-                ch.pipeline().addLast("logger", new LoggingHandler());
-                ch.pipeline().addLast("packetDecoder", new UdpPacketToByteBuf());
-                ch.pipeline().addLast("buffer", new MetricsBufferDecoder());
-                ch.pipeline().addLast("frame", new DelimiterBasedFrameDecoder(65536, true, Delimiters.lineDelimiter()));
-                ch.pipeline().addLast("putDecoder", new UdpDecoder());
-                ch.pipeline().addLast("udpRelayHandler", new UdpRelayHandler(metricResolver, udpClientPool));
-            }
-        };
-    }
-
-    protected ChannelHandler setupTcpChannel(MetricResolver metricResolver, List<TcpClientPool> tcpClientPools) {
-        return new ChannelInitializer<SocketChannel>() {
-
-            @Override
-            protected void initChannel(SocketChannel ch) throws Exception {
-                ch.pipeline().addLast("buffer", new MetricsBufferDecoder());
-                ch.pipeline().addLast("frame", new DelimiterBasedFrameDecoder(65536, true, Delimiters.lineDelimiter()));
-                ch.pipeline().addLast("putDecoder", new TcpDecoder());
-                ch.pipeline().addLast("tcpRelayHandler", new TcpRelayHandler(metricResolver, tcpClientPools));
-                ch.pipeline().addLast("versionHandler", new TcpVersionHandler());
-            }
-        };
-    }
-
-    protected ChannelHandler setupWSChannel(BalancerWebsocketProperties balancerWebsocketProperties, SecurityProperties securityProperties, SslContext sslCtx,
-                    MetricResolver metricResolver, WsClientPool wsClientPool) {
-        return new ChannelInitializer<SocketChannel>() {
-
-            @Override
-            protected void initChannel(SocketChannel ch) throws Exception {
-                ch.pipeline().addLast("ssl", sslCtx.newHandler(ch.alloc()));
-                ch.pipeline().addLast("httpServer", new HttpServerCodec());
-                ch.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
-                ch.pipeline().addLast("sessionExtractor", new WebSocketFullRequestHandler(authenticationService));
-
-                ch.pipeline().addLast("idle-handler", new IdleStateHandler(balancerWebsocketProperties.getTimeout(), 0, 0));
-                ch.pipeline().addLast("ws-protocol", new WebSocketServerProtocolHandler(WS_PATH, null, true, 65536, false, true));
-                ch.pipeline().addLast("wsDecoder", new WebSocketRequestDecoder(authenticationService, securityProperties));
-                ch.pipeline().addLast("httpRelay", new WsRelayHandler(balancerWebsocketProperties, metricResolver, wsClientPool));
-                ch.pipeline().addLast("error", new WSTimelyExceptionHandler());
-            }
-        };
-    }
-}

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

@@ -1,10 +1,173 @@
 package timely.balancer.configuration;
 
-import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.context.annotation.Configuration;
+import javax.validation.Valid;
 
-@Configuration
-@EnableConfigurationProperties({BalancerProperties.class, BalancerServerProperties.class, BalancerWebsocketProperties.class, BalancerHttpProperties.class})
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.NestedConfigurationProperty;
+import org.springframework.stereotype.Component;
+import org.springframework.validation.annotation.Validated;
+import timely.configuration.Cache;
+
+@Validated
+@Component
+@ConfigurationProperties(prefix = "timely-balancer")
 public class BalancerConfiguration {
 
+    @Valid
+    @NestedConfigurationProperty
+    private BalancerServer server = new BalancerServer();
+
+    @Valid
+    @NestedConfigurationProperty
+    private BalancerHttp http = new BalancerHttp();
+
+    @Valid
+    @NestedConfigurationProperty
+    private BalancerWebsocket websocket = new BalancerWebsocket();
+
+    @Valid
+    @NestedConfigurationProperty
+    private ZooKeeper zooKeeper = new ZooKeeper();
+
+    @Valid
+    @NestedConfigurationProperty
+    private BalancerSecurity security = new BalancerSecurity();
+
+    @Valid
+    @NestedConfigurationProperty
+    private Cache cache = new Cache();
+
+    private boolean loginRequired = false;
+
+    private String assignmentFile;
+
+    private Double controlBandPercentage = 0.05;
+
+    private long checkServerHealthInterval = 10000;
+
+    private int serverFailuresBeforeDown = 3;
+
+    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;
+    }
+
+    public void setLoginRequired(boolean loginRequired) {
+        this.loginRequired = loginRequired;
+    }
+
+    public boolean isLoginRequired() {
+        return loginRequired;
+    }
+
+    public String getAssignmentFile() {
+        return assignmentFile;
+    }
+
+    public void setAssignmentFile(String assignmentFile) {
+        this.assignmentFile = assignmentFile;
+    }
+
+    public void setControlBandPercentage(Double controlBandPercentage) {
+        this.controlBandPercentage = controlBandPercentage;
+    }
+
+    public Double getControlBandPercentage() {
+        return controlBandPercentage;
+    }
+
+    public BalancerSecurity getSecurity() {
+        return security;
+    }
+
+    public BalancerServer getServer() {
+        return server;
+    }
+
+    public BalancerHttp getHttp() {
+        return http;
+    }
+
+    public BalancerWebsocket getWebsocket() {
+        return websocket;
+    }
+
+    public Cache getCache() {
+        return cache;
+    }
+
+    public void setCheckServerHealthInterval(long checkServerHealthInterval) {
+        this.checkServerHealthInterval = checkServerHealthInterval;
+    }
+
+    public long getCheckServerHealthInterval() {
+        return checkServerHealthInterval;
+    }
+
+    public int getServerFailuresBeforeDown() {
+        return serverFailuresBeforeDown;
+    }
+
+    public void setServerFailuresBeforeDown(int serverFailuresBeforeDown) {
+        this.serverFailuresBeforeDown = serverFailuresBeforeDown;
+    }
+
+    public int getServerSuccessesBeforeUp() {
+        return serverSuccessesBeforeUp;
+    }
+
+    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/configuration/BalancerHttp.java

@@ -4,15 +4,10 @@ import javax.validation.Valid;
 
 import org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig;
 import org.apache.http.impl.client.CloseableHttpClient;
-import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.NestedConfigurationProperty;
-import org.springframework.cloud.context.config.annotation.RefreshScope;
+import timely.configuration.Http;
 
-import timely.common.configuration.HttpProperties;
-
-@RefreshScope
-@ConfigurationProperties(prefix = "timely.balancer.http")
-public class BalancerHttpProperties extends HttpProperties {
+public class BalancerHttp extends Http {
 
     @Valid
     @NestedConfigurationProperty

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

@@ -1,71 +0,0 @@
-package timely.balancer.configuration;
-
-import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.stereotype.Component;
-import org.springframework.validation.annotation.Validated;
-
-@Validated
-@Component
-@ConfigurationProperties(prefix = "timely.balancer")
-public class BalancerProperties {
-
-    private boolean loginRequired = false;
-
-    private String assignmentFile;
-
-    private Double controlBandPercentage = 0.05;
-
-    private long checkServerHealthInterval = 10000;
-
-    private int serverFailuresBeforeDown = 3;
-
-    private int serverSuccessesBeforeUp = 3;
-
-    public void setLoginRequired(boolean loginRequired) {
-        this.loginRequired = loginRequired;
-    }
-
-    public boolean isLoginRequired() {
-        return loginRequired;
-    }
-
-    public String getAssignmentFile() {
-        return assignmentFile;
-    }
-
-    public void setAssignmentFile(String assignmentFile) {
-        this.assignmentFile = assignmentFile;
-    }
-
-    public void setControlBandPercentage(Double controlBandPercentage) {
-        this.controlBandPercentage = controlBandPercentage;
-    }
-
-    public Double getControlBandPercentage() {
-        return controlBandPercentage;
-    }
-
-    public void setCheckServerHealthInterval(long checkServerHealthInterval) {
-        this.checkServerHealthInterval = checkServerHealthInterval;
-    }
-
-    public long getCheckServerHealthInterval() {
-        return checkServerHealthInterval;
-    }
-
-    public int getServerFailuresBeforeDown() {
-        return serverFailuresBeforeDown;
-    }
-
-    public void setServerFailuresBeforeDown(int serverFailuresBeforeDown) {
-        this.serverFailuresBeforeDown = serverFailuresBeforeDown;
-    }
-
-    public int getServerSuccessesBeforeUp() {
-        return serverSuccessesBeforeUp;
-    }
-
-    public void setServerSuccessesBeforeUp(int serverSuccessesBeforeUp) {
-        this.serverSuccessesBeforeUp = serverSuccessesBeforeUp;
-    }
-}

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

@@ -0,0 +1,17 @@
+package timely.balancer.configuration;
+
+import javax.validation.Valid;
+
+import org.springframework.boot.context.properties.NestedConfigurationProperty;
+import timely.configuration.Security;
+
+public class BalancerSecurity extends Security {
+
+    @Valid
+    @NestedConfigurationProperty
+    private ClientSsl clientSsl = new ClientSsl();
+
+    public ClientSsl getClientSsl() {
+        return clientSsl;
+    }
+}

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

@@ -3,17 +3,12 @@ package timely.balancer.configuration;
 import javax.validation.Valid;
 
 import org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig;
-import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.NestedConfigurationProperty;
-import org.springframework.cloud.context.config.annotation.RefreshScope;
-
 import timely.client.tcp.TcpClient;
 import timely.client.udp.UdpClient;
-import timely.common.configuration.ServerProperties;
+import timely.configuration.Server;
 
-@RefreshScope
-@ConfigurationProperties(prefix = "timely.balancer.server")
-public class BalancerServerProperties extends ServerProperties {
+public class BalancerServer extends Server {
 
     @Valid
     @NestedConfigurationProperty

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

@@ -3,16 +3,11 @@ package timely.balancer.configuration;
 import javax.validation.Valid;
 
 import org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig;
-import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.NestedConfigurationProperty;
-import org.springframework.cloud.context.config.annotation.RefreshScope;
-
 import timely.client.websocket.subscription.WebSocketSubscriptionClient;
-import timely.common.configuration.WebsocketProperties;
+import timely.configuration.Websocket;
 
-@RefreshScope
-@ConfigurationProperties(prefix = "timely.balancer.websocket")
-public class BalancerWebsocketProperties extends WebsocketProperties {
+public class BalancerWebsocket extends Websocket {
 
     @Valid
     @NestedConfigurationProperty
@@ -22,7 +17,7 @@ public class BalancerWebsocketProperties extends WebsocketProperties {
         return wsClientPool;
     }
 
-    private int incomingBufferSize = 50000000;
+    public int incomingBufferSize = 50000000;
 
     public int getIncomingBufferSize() {
         return incomingBufferSize;

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

@@ -0,0 +1,77 @@
+package timely.balancer.configuration;
+
+public class ClientSsl {
+
+    private String keyFile;
+    private String keyType;
+    private String keyPassword;
+    private String trustStoreFile;
+    private String trustStoreType;
+    private String trustStorePassword;
+    private boolean useClientCert = true;
+    private boolean hostVerificationEnabled = true;
+
+    public String getKeyFile() {
+        return keyFile;
+    }
+
+    public void setKeyFile(String keyFile) {
+        this.keyFile = keyFile;
+    }
+
+    public String getKeyType() {
+        return keyType;
+    }
+
+    public void setKeyType(String keyType) {
+        this.keyType = keyType;
+    }
+
+    public String getKeyPassword() {
+        return keyPassword;
+    }
+
+    public void setKeyPassword(String keyPassword) {
+        this.keyPassword = keyPassword;
+    }
+
+    public String getTrustStoreFile() {
+        return trustStoreFile;
+    }
+
+    public void setTrustStoreFile(String trustStoreFile) {
+        this.trustStoreFile = trustStoreFile;
+    }
+
+    public String getTrustStoreType() {
+        return trustStoreType;
+    }
+
+    public void setTrustStoreType(String trustStoreType) {
+        this.trustStoreType = trustStoreType;
+    }
+
+    public String getTrustStorePassword() {
+        return trustStorePassword;
+    }
+
+    public void setTrustStorePassword(String trustStorePassword) {
+        this.trustStorePassword = trustStorePassword;
+    }
+
+    public void setUseClientCert(boolean useClientCert) {
+        this.useClientCert = useClientCert;
+    }
+
+    public boolean isUseClientCert() {
+        return useClientCert;
+    }
+
+    public void setHostVerificationEnabled(boolean hostVerificationEnabled) {
+        this.hostVerificationEnabled = hostVerificationEnabled;
+    }
+
+    public boolean isHostVerificationEnabled() {
+        return hostVerificationEnabled;
+    }
+}

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

@@ -2,5 +2,6 @@ package timely.balancer.configuration;
 
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 
-@SpringBootApplication(scanBasePackageClasses = {SpringBootstrap.class})
-public class SpringBootstrap {}
+@SpringBootApplication(scanBasePackageClasses = { SpringBootstrap.class })
+public class SpringBootstrap {
+}

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

@@ -1,13 +1,9 @@
-package timely.common.configuration;
+package timely.balancer.configuration;
 
 import javax.validation.constraints.NotBlank;
 
-import org.springframework.boot.context.properties.ConfigurationProperties;
+public class ZooKeeper {
 
-@ConfigurationProperties(prefix = "timely.zookeeper")
-public class ZookeeperProperties {
-
-    @NotBlank
     private String servers;
 
     @NotBlank

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

@@ -5,13 +5,12 @@ import org.apache.commons.lang3.builder.HashCodeBuilder;
 import org.apache.commons.lang3.builder.ToStringBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import timely.balancer.ArrivalRate;
-import timely.balancer.configuration.BalancerProperties;
+import timely.balancer.configuration.BalancerConfiguration;
 
 public class TimelyBalancedHost {
 
-    private static final Logger log = LoggerFactory.getLogger(TimelyBalancedHost.class);
+    private static final Logger LOG = LoggerFactory.getLogger(TimelyBalancedHost.class);
 
     private String host;
     private int tcpPort;
@@ -19,7 +18,7 @@ public class TimelyBalancedHost {
     private int wsPort;
     private int udpPort;
     private boolean isUp = true;
-    private BalancerProperties balancerProperties;
+    private BalancerConfiguration balancerConfig;
     private int failures = 0;
     private int successes = 0;
     private ArrivalRate arrivalRate = null;
@@ -28,7 +27,8 @@ public class TimelyBalancedHost {
 
     }
 
-    private TimelyBalancedHost(String host, int tcpPort, int httpPort, int wsPort, int udpPort, ArrivalRate arrivalRate) {
+    private TimelyBalancedHost(String host, int tcpPort, int httpPort, int wsPort, int udpPort,
+            ArrivalRate arrivalRate) {
         this.host = host;
         this.tcpPort = tcpPort;
         this.httpPort = httpPort;
@@ -37,12 +37,17 @@ public class TimelyBalancedHost {
         this.arrivalRate = arrivalRate;
     }
 
-    static public TimelyBalancedHost of(String host, int tcpPort, int httpPort, int wsPort, int udpPort, ArrivalRate arrivalRate) {
+    static public TimelyBalancedHost of(String host, int tcpPort, int httpPort, int wsPort, int udpPort,
+            ArrivalRate arrivalRate) {
         return new TimelyBalancedHost(host, tcpPort, httpPort, wsPort, udpPort, arrivalRate);
     }
 
-    public void setBalancerProperties(BalancerProperties balancerProperties) {
-        this.balancerProperties = balancerProperties;
+    static public TimelyBalancedHost of(String host, int basePort) {
+        return new TimelyBalancedHost(host, basePort, basePort + 1, basePort + 2, basePort + 3, null);
+    }
+
+    public void setBalancerConfig(BalancerConfiguration balancerConfig) {
+        this.balancerConfig = balancerConfig;
     }
 
     synchronized public boolean isUp() {
@@ -50,38 +55,38 @@ public class TimelyBalancedHost {
     }
 
     public void reportSuccess() {
-        int serverSuccessesBeforeUp = balancerProperties.getServerSuccessesBeforeUp();
+        int serverSuccessesBeforeUp = balancerConfig.getServerSuccessesBeforeUp();
         String h = host;
         int p = tcpPort;
         synchronized (this) {
-            if (log.isTraceEnabled()) {
-                log.trace("success reported host:{} port:{} isUp:{} successes:{} serverSuccessesBeforeUp:{}", h, p, isUp, (successes + 1),
-                                serverSuccessesBeforeUp);
+            if (LOG.isTraceEnabled()) {
+                LOG.trace("success reported host:{} port:{} isUp:{} successes:{} serverSuccessesBeforeUp:{}", h, p,
+                        isUp, (successes + 1), serverSuccessesBeforeUp);
             }
             if (!isUp) {
                 if (++successes >= serverSuccessesBeforeUp) {
                     isUp = true;
                     successes = 0;
-                    log.info("host up host:{} port:{}", h, p);
+                    LOG.info("host up host:{} port:{}", h, p);
                 }
             }
         }
     }
 
     public void reportFailure() {
-        int serverFailuresBeforeDown = balancerProperties.getServerFailuresBeforeDown();
+        int serverFailuresBeforeDown = balancerConfig.getServerFailuresBeforeDown();
         String h = host;
         int p = tcpPort;
         synchronized (this) {
-            if (log.isTraceEnabled()) {
-                log.trace("failure reported host:{} port:{} isUp:{} failures:{} serverFailuresBeforeUp:{}", h, p, isUp, (failures + 1),
-                                serverFailuresBeforeDown);
+            if (LOG.isTraceEnabled()) {
+                LOG.trace("failure reported host:{} port:{} isUp:{} failures:{} serverFailuresBeforeUp:{}", h, p, isUp,
+                        (failures + 1), serverFailuresBeforeDown);
             }
             if (isUp) {
                 if (++failures >= serverFailuresBeforeDown) {
                     isUp = false;
                     failures = 0;
-                    log.info("host down host:{} port:{}", h, p);
+                    LOG.info("host down host:{} port:{}", h, p);
                 }
             }
         }
@@ -179,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/connection/http/HttpClientFactory.java

@@ -6,28 +6,24 @@ import org.apache.commons.pool2.KeyedPooledObjectFactory;
 import org.apache.commons.pool2.PooledObject;
 import org.apache.commons.pool2.impl.DefaultPooledObject;
 import org.apache.http.impl.client.CloseableHttpClient;
-
+import timely.balancer.configuration.BalancerSecurity;
 import timely.balancer.connection.TimelyBalancedHost;
-import timely.common.configuration.SecurityProperties;
-import timely.common.configuration.SslClientProperties;
 
-public class HttpClientFactory implements KeyedPooledObjectFactory<TimelyBalancedHost,CloseableHttpClient> {
+public class HttpClientFactory implements KeyedPooledObjectFactory<TimelyBalancedHost, CloseableHttpClient> {
 
     private final SSLContext sslContext;
-    private final SecurityProperties security;
-    private SslClientProperties sslClientProperties;
+    private final BalancerSecurity security;
 
-    public HttpClientFactory(SecurityProperties security, SslClientProperties sslClientProperties, SSLContext sslContext) {
+    public HttpClientFactory(BalancerSecurity security, SSLContext sslContext) {
         this.security = security;
         this.sslContext = sslContext;
-        this.sslClientProperties = sslClientProperties;
     }
 
     @Override
     public PooledObject<CloseableHttpClient> makeObject(TimelyBalancedHost k) throws Exception {
         // disable cookie management because we are sharing connections
-        return new DefaultPooledObject<>(timely.client.http.HttpClient.get(this.sslContext, null, sslClientProperties.isHostVerificationEnabled(),
-                        sslClientProperties.isUseClientCert()));
+        return new DefaultPooledObject<>(timely.client.http.HttpClient.get(this.sslContext, null,
+                security.getClientSsl().isHostVerificationEnabled(), security.getClientSsl().isUseClientCert()));
     }
 
     @Override

balancer/src/main/java/timely/balancer/connection/http/HttpClientPool.java

@@ -4,15 +4,13 @@ import javax.net.ssl.SSLContext;
 
 import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
 import org.apache.http.impl.client.CloseableHttpClient;
-
-import timely.balancer.configuration.BalancerHttpProperties;
+import timely.balancer.configuration.BalancerHttp;
+import timely.balancer.configuration.BalancerSecurity;
 import timely.balancer.connection.TimelyBalancedHost;
-import timely.common.configuration.SecurityProperties;
-import timely.common.configuration.SslClientProperties;
 
-public class HttpClientPool extends GenericKeyedObjectPool<TimelyBalancedHost,CloseableHttpClient> {
+public class HttpClientPool extends GenericKeyedObjectPool<TimelyBalancedHost, CloseableHttpClient> {
 
-    public HttpClientPool(SecurityProperties securityProperties, SslClientProperties sslClientProperties, BalancerHttpProperties http, SSLContext sslContext) {
-        super(new HttpClientFactory(securityProperties, sslClientProperties, sslContext), http.getHttpClientPool());
+    public HttpClientPool(BalancerSecurity balancerSecurity, BalancerHttp http, SSLContext sslContext) {
+        super(new HttpClientFactory(balancerSecurity, sslContext), http.getHttpClientPool());
     }
 }

balancer/src/main/java/timely/balancer/connection/tcp/TcpClientFactory.java

@@ -5,11 +5,10 @@ import java.io.IOException;
 import org.apache.commons.pool2.KeyedPooledObjectFactory;
 import org.apache.commons.pool2.PooledObject;
 import org.apache.commons.pool2.impl.DefaultPooledObject;
-
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.client.tcp.TcpClient;
 
-public class TcpClientFactory implements KeyedPooledObjectFactory<TimelyBalancedHost,TcpClient> {
+public class TcpClientFactory implements KeyedPooledObjectFactory<TimelyBalancedHost, TcpClient> {
 
     private int bufferSize;
 

balancer/src/main/java/timely/balancer/connection/tcp/TcpClientPool.java

@@ -1,14 +1,13 @@
 package timely.balancer.connection.tcp;
 
 import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
-
-import timely.balancer.configuration.BalancerServerProperties;
+import timely.balancer.configuration.BalancerConfiguration;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.client.tcp.TcpClient;
 
-public class TcpClientPool extends GenericKeyedObjectPool<TimelyBalancedHost,TcpClient> {
+public class TcpClientPool extends GenericKeyedObjectPool<TimelyBalancedHost, TcpClient> {
 
-    public TcpClientPool(BalancerServerProperties balancerServerProperties) {
-        super(new TcpClientFactory(balancerServerProperties.getTcpBufferSize()), balancerServerProperties.getTcpClientPool());
+    public TcpClientPool(BalancerConfiguration config) {
+        super(new TcpClientFactory(config.getServer().getTcpBufferSize()), config.getServer().getTcpClientPool());
     }
-}
+}

balancer/src/main/java/timely/balancer/connection/udp/UdpClientFactory.java

@@ -5,13 +5,13 @@ import java.io.IOException;
 import org.apache.commons.pool2.KeyedPooledObjectFactory;
 import org.apache.commons.pool2.PooledObject;
 import org.apache.commons.pool2.impl.DefaultPooledObject;
-
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.client.udp.UdpClient;
 
-public class UdpClientFactory implements KeyedPooledObjectFactory<TimelyBalancedHost,UdpClient> {
+public class UdpClientFactory implements KeyedPooledObjectFactory<TimelyBalancedHost, UdpClient> {
 
-    public UdpClientFactory() {}
+    public UdpClientFactory() {
+    }
 
     @Override
     public PooledObject<UdpClient> makeObject(TimelyBalancedHost k) throws Exception {

balancer/src/main/java/timely/balancer/connection/udp/UdpClientPool.java

@@ -1,14 +1,13 @@
 package timely.balancer.connection.udp;
 
 import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
-
-import timely.balancer.configuration.BalancerServerProperties;
+import timely.balancer.configuration.BalancerConfiguration;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.client.udp.UdpClient;
 
-public class UdpClientPool extends GenericKeyedObjectPool<TimelyBalancedHost,UdpClient> {
+public class UdpClientPool extends GenericKeyedObjectPool<TimelyBalancedHost, UdpClient> {
 
-    public UdpClientPool(BalancerServerProperties serverProperties) {
-        super(new UdpClientFactory(), serverProperties.getUdpClientPool());
+    public UdpClientPool(BalancerConfiguration config) {
+        super(new UdpClientFactory(), config.getServer().getUdpClientPool());
     }
 }

balancer/src/main/java/timely/balancer/connection/ws/WsClientFactory.java

@@ -5,35 +5,28 @@ import javax.net.ssl.SSLContext;
 import org.apache.commons.pool2.KeyedPooledObjectFactory;
 import org.apache.commons.pool2.PooledObject;
 import org.apache.commons.pool2.impl.DefaultPooledObject;
-
-import timely.balancer.configuration.BalancerProperties;
-import timely.balancer.configuration.BalancerWebsocketProperties;
+import timely.balancer.configuration.BalancerConfiguration;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.client.websocket.subscription.WebSocketSubscriptionClient;
-import timely.common.configuration.SslClientProperties;
 
-public class WsClientFactory implements KeyedPooledObjectFactory<TimelyBalancedHost,WebSocketSubscriptionClient> {
+public class WsClientFactory implements KeyedPooledObjectFactory<TimelyBalancedHost, WebSocketSubscriptionClient> {
 
     private final SSLContext sslContext;
-    private BalancerProperties balancerProperties;
-    private BalancerWebsocketProperties balancerWebsocketProperties;
-    private SslClientProperties sslClientProperties;
+    private BalancerConfiguration balancerConfig;
 
-    public WsClientFactory(BalancerProperties balancerProperties, BalancerWebsocketProperties balancerWebsocketProperties,
-                    SslClientProperties sslClientProperties, SSLContext sslContext) {
-        this.balancerProperties = balancerProperties;
-        this.balancerWebsocketProperties = balancerWebsocketProperties;
-        this.sslClientProperties = sslClientProperties;
+    public WsClientFactory(BalancerConfiguration balancerConfig, SSLContext sslContext) {
+        this.balancerConfig = balancerConfig;
         this.sslContext = sslContext;
     }
 
     @Override
     public PooledObject<WebSocketSubscriptionClient> makeObject(TimelyBalancedHost k) throws Exception {
 
-        int bufferSize = balancerWebsocketProperties.getIncomingBufferSize();
-        WebSocketSubscriptionClient client = new WebSocketSubscriptionClient(sslContext, k.getHost(), k.getHttpPort(), k.getWsPort(),
-                        sslClientProperties.isUseClientCert(), balancerProperties.isLoginRequired(), sslClientProperties.isHostVerificationEnabled(),
-                        bufferSize);
+        int bufferSize = balancerConfig.getWebsocket().getIncomingBufferSize();
+        WebSocketSubscriptionClient client = new WebSocketSubscriptionClient(sslContext, k.getHost(), k.getHttpPort(),
+                k.getWsPort(), balancerConfig.getSecurity().getClientSsl().isUseClientCert(),
+                balancerConfig.isLoginRequired(), "", "",
+                balancerConfig.getSecurity().getClientSsl().isHostVerificationEnabled(), bufferSize);
         return new DefaultPooledObject<>(client);
     }
 

balancer/src/main/java/timely/balancer/connection/ws/WsClientPool.java

@@ -3,18 +3,13 @@ package timely.balancer.connection.ws;
 import javax.net.ssl.SSLContext;
 
 import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
-
-import timely.balancer.configuration.BalancerProperties;
-import timely.balancer.configuration.BalancerWebsocketProperties;
+import timely.balancer.configuration.BalancerConfiguration;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.client.websocket.subscription.WebSocketSubscriptionClient;
-import timely.common.configuration.SslClientProperties;
 
-public class WsClientPool extends GenericKeyedObjectPool<TimelyBalancedHost,WebSocketSubscriptionClient> {
+public class WsClientPool extends GenericKeyedObjectPool<TimelyBalancedHost, WebSocketSubscriptionClient> {
 
-    public WsClientPool(BalancerProperties balancerProperties, BalancerWebsocketProperties balancerWebsocketProperties, SslClientProperties sslClientProperties,
-                    SSLContext sslContext) {
-        super(new WsClientFactory(balancerProperties, balancerWebsocketProperties, sslClientProperties, sslContext),
-                        balancerWebsocketProperties.getWsClientPool());
+    public WsClientPool(BalancerConfiguration balancerConfig, SSLContext sslContext) {
+        super(new WsClientFactory(balancerConfig, sslContext), balancerConfig.getWebsocket().getWsClientPool());
     }
 }

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

@@ -0,0 +1,12 @@
+package timely.balancer.healthcheck;
+
+import java.util.List;
+
+import timely.balancer.connection.TimelyBalancedHost;
+
+public interface HealthChecker {
+
+    public void setTimelyHosts(List<TimelyBalancedHost> timelyHosts);
+
+    public void close() throws Exception;
+}

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

@@ -1,4 +1,4 @@
-package timely.balancer.component;
+package timely.balancer.healthcheck;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -6,39 +6,20 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.stereotype.Component;
-
-import timely.balancer.configuration.BalancerProperties;
-import timely.balancer.configuration.BalancerServerProperties;
+import timely.balancer.configuration.BalancerConfiguration;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.balancer.connection.tcp.TcpClientPool;
-import timely.balancer.healthcheck.TCPServerHealthCheck;
 
-@Component
-@ConditionalOnMissingBean(HealthChecker.class)
-public class HealthChecker {
+public class TcpHealthChecker implements HealthChecker {
 
-    private static final Logger log = LoggerFactory.getLogger(HealthChecker.class);
+    private static final Logger LOG = LoggerFactory.getLogger(TcpHealthChecker.class);
+    private List<TimelyBalancedHost> timelyHosts = new ArrayList<>();
     private ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
-    private TCPServerHealthCheck tcpServerHealthCheck;
-    private BalancerProperties balancerProperties;
-    private BalancerServerProperties balancerServerProperties;
-    protected List<TimelyBalancedHost> timelyHosts = new ArrayList<>();
+    final private TCPServerHealthCheck tcpServerHealthCheck;
 
-    public HealthChecker(BalancerProperties balancerProperties, BalancerServerProperties balancerServerProperties) {
-        this.balancerProperties = balancerProperties;
-        this.balancerServerProperties = balancerServerProperties;
-    }
-
-    @PostConstruct
-    public void setup() {
-        TcpClientPool tcpClientPool = new TcpClientPool(balancerServerProperties);
+    public TcpHealthChecker(BalancerConfiguration config, TcpClientPool tcpClientPool) {
         this.tcpServerHealthCheck = new TCPServerHealthCheck(tcpClientPool);
         this.executorService.scheduleAtFixedRate(() -> {
             synchronized (timelyHosts) {
@@ -50,15 +31,22 @@ public class HealthChecker {
                             h.reportFailure();
                         }
                     } catch (Exception e) {
-                        log.error(e.getMessage(), e);
+                        LOG.error(e.getMessage(), e);
                         h.reportFailure();
                     }
                 }
             }
-        }, 0, balancerProperties.getCheckServerHealthInterval(), TimeUnit.MILLISECONDS);
+        }, 0, config.getCheckServerHealthInterval(), TimeUnit.MILLISECONDS);
+    }
+
+    @Override
+    public void setTimelyHosts(List<TimelyBalancedHost> timelyHosts) {
+        synchronized (timelyHosts) {
+            this.timelyHosts.clear();
+            this.timelyHosts.addAll(timelyHosts);
+        }
     }
 
-    @PreDestroy
     public void close() throws Exception {
         this.executorService.shutdown();
         try {
@@ -71,11 +59,4 @@ public class HealthChecker {
             }
         }
     }
-
-    public void setTimelyHosts(List<TimelyBalancedHost> timelyHosts) {
-        synchronized (timelyHosts) {
-            this.timelyHosts.clear();
-            this.timelyHosts.addAll(timelyHosts);
-        }
-    }
 }

balancer/src/main/java/timely/balancer/netty/http/HttpRelayHandler.java

@@ -11,6 +11,17 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 
+import com.google.common.collect.Multimap;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpVersion;
+import io.netty.util.ReferenceCountUtil;
 import org.apache.http.Header;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpGet;
@@ -22,35 +33,22 @@ import org.apache.http.message.BasicHeader;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.util.StreamUtils;
-
-import com.google.common.collect.Multimap;
-
-import io.netty.buffer.Unpooled;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.SimpleChannelInboundHandler;
-import io.netty.handler.codec.http.DefaultFullHttpResponse;
-import io.netty.handler.codec.http.FullHttpRequest;
-import io.netty.handler.codec.http.FullHttpResponse;
-import io.netty.handler.codec.http.HttpMethod;
-import io.netty.handler.codec.http.HttpResponseStatus;
-import io.netty.handler.codec.http.HttpVersion;
-import io.netty.util.ReferenceCountUtil;
 import timely.api.request.AuthenticatedRequest;
-import timely.api.request.HttpRequest;
 import timely.api.request.MetricRequest;
+import timely.api.request.timeseries.HttpRequest;
 import timely.api.request.timeseries.QueryRequest;
 import timely.api.response.TimelyException;
-import timely.auth.TimelyAuthenticationToken;
 import timely.auth.util.HttpHeaderUtils;
 import timely.auth.util.ProxiedEntityUtils;
-import timely.balancer.component.MetricResolver;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.balancer.connection.http.HttpClientPool;
+import timely.balancer.resolver.MetricResolver;
 import timely.netty.http.TimelyHttpHandler;
+import timely.netty.http.auth.TimelyAuthenticationToken;
 
 public class HttpRelayHandler extends SimpleChannelInboundHandler<HttpRequest> implements TimelyHttpHandler {
 
-    private static final Logger log = LoggerFactory.getLogger(HttpRelayHandler.class);
+    private static final Logger LOG = LoggerFactory.getLogger(HttpRelayHandler.class);
     private final HttpClientPool httpClientPool;
     private MetricResolver metricResolver;
 
@@ -90,7 +88,7 @@ public class HttpRelayHandler extends SimpleChannelInboundHandler<HttpRequest> i
                 }
                 client = httpClientPool.borrowObject(k);
 
-                Multimap<String,String> headers = HttpHeaderUtils.toMultimap(request.headers());
+                Multimap<String, String> headers = HttpHeaderUtils.toMultimap(request.headers());
                 if (msg instanceof AuthenticatedRequest) {
                     AuthenticatedRequest authenticatedRequest = (AuthenticatedRequest) msg;
                     TimelyAuthenticationToken token = authenticatedRequest.getToken();
@@ -102,7 +100,7 @@ public class HttpRelayHandler extends SimpleChannelInboundHandler<HttpRequest> i
                 }
 
                 List<Header> relayedHeaderList = new ArrayList<>();
-                for (Map.Entry<String,String> h : headers.entries()) {
+                for (Map.Entry<String, String> h : headers.entries()) {
                     if (!h.getKey().equals(CONTENT_LENGTH.toString())) {
                         relayedHeaderList.add(new BasicHeader(h.getKey(), h.getValue()));
                     }
@@ -132,30 +130,33 @@ public class HttpRelayHandler extends SimpleChannelInboundHandler<HttpRequest> i
             } catch (Exception e) {
                 String message = e.getMessage();
                 if (message == null) {
-                    log.error("", e);
+                    LOG.error("", e);
                 } else {
                     if (message.contains("No matching tags")) {
-                        log.trace(message);
+                        LOG.trace(message);
                     } else {
-                        log.error(message, e);
+                        LOG.error(message, e);
                     }
                 }
-                this.sendHttpError(ctx, new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), e.getMessage(), e.getLocalizedMessage(), e));
+                this.sendHttpError(ctx, new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
+                        e.getMessage(), e.getLocalizedMessage(), e));
                 return;
             } finally {
-                if (request != null) {
+                if (request != null && request instanceof AutoCloseable) {
                     ReferenceCountUtil.release(request);
                 }
             }
             FullHttpResponse response = null;
             if (relayedResponse == null) {
-                this.sendHttpError(ctx, new TimelyException(HttpResponseStatus.METHOD_NOT_ALLOWED.code(), "Method not allowed", "Method not allowed"));
+                this.sendHttpError(ctx, new TimelyException(HttpResponseStatus.METHOD_NOT_ALLOWED.code(),
+                        "Method not allowed", "Method not allowed"));
             } else {
                 ByteArrayOutputStream baos = new MyByteArrayOutputStream();
                 StreamUtils.copy(relayedResponse.getEntity().getContent(), baos);
 
-                response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.valueOf(relayedResponse.getStatusLine().getStatusCode()),
-                                Unpooled.copiedBuffer(baos.toByteArray()));
+                response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
+                        HttpResponseStatus.valueOf(relayedResponse.getStatusLine().getStatusCode()),
+                        Unpooled.copiedBuffer(baos.toByteArray()));
                 for (Header h : relayedResponse.getAllHeaders()) {
                     response.headers().add(h.getName(), h.getValue());
                 }
@@ -169,14 +170,15 @@ public class HttpRelayHandler extends SimpleChannelInboundHandler<HttpRequest> i
             if (client != null && k != null) {
                 httpClientPool.returnObject(k, client);
             } else {
-                log.error("NOT RETURNING CONNECTION! " + msg.getHttpRequest().uri());
+                LOG.error("NOT RETURNING CONNECTION! " + msg.getHttpRequest().uri());
             }
         }
     }
 
     static public class MyByteArrayOutputStream extends ByteArrayOutputStream {
 
-        public MyByteArrayOutputStream() {}
+        public MyByteArrayOutputStream() {
+        }
 
         public MyByteArrayOutputStream(int size) {
             super(size);

balancer/src/main/java/timely/balancer/netty/tcp/TcpRelayHandler.java

@@ -4,33 +4,33 @@ import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
 
-import org.apache.commons.lang3.tuple.Pair;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
+import org.apache.commons.lang3.tuple.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import timely.api.request.MetricRequest;
 import timely.api.request.TcpRequest;
-import timely.balancer.component.MetricResolver;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.balancer.connection.tcp.TcpClientPool;
+import timely.balancer.resolver.BalancedMetricResolver;
+import timely.balancer.resolver.MetricResolver;
 import timely.client.tcp.TcpClient;
 import timely.netty.Constants;
 
 public class TcpRelayHandler extends SimpleChannelInboundHandler<TcpRequest> {
 
-    private static final Logger log = LoggerFactory.getLogger(TcpRelayHandler.class);
+    private static final Logger LOG = LoggerFactory.getLogger(TcpRelayHandler.class);
     private static final String LOG_ERR_MSG = "Error storing put metric: {}";
     private static final String ERR_MSG = "Error storing put metric: ";
 
-    private MetricResolver metricResolver;
+    private BalancedMetricResolver metricResolver;
     private List<TcpClientPool> tcpClientPools;
 
     public TcpRelayHandler(MetricResolver metricResolver, List<TcpClientPool> tcpClientPools) {
-        this.metricResolver = metricResolver;
+        this.metricResolver = (BalancedMetricResolver) metricResolver;
         this.tcpClientPools = tcpClientPools;
     }
 
@@ -41,10 +41,10 @@ public class TcpRelayHandler extends SimpleChannelInboundHandler<TcpRequest> {
 
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, TcpRequest msg) throws Exception {
-        log.trace("Received {}", msg);
+        LOG.trace("Received {}", msg);
         try {
             String line;
-            Pair<TimelyBalancedHost,TcpClient> keyClientPair = null;
+            Pair<TimelyBalancedHost, TcpClient> keyClientPair = null;
             try {
                 if (msg instanceof MetricRequest) {
                     String metricName = ((MetricRequest) msg).getMetric().getName();
@@ -63,27 +63,29 @@ public class TcpRelayHandler extends SimpleChannelInboundHandler<TcpRequest> {
             }
 
         } catch (IOException e) {
-            log.error(LOG_ERR_MSG, msg, e);
-            ChannelFuture cf = ctx.writeAndFlush(Unpooled.copiedBuffer((ERR_MSG + e.getMessage() + "\n").getBytes(StandardCharsets.UTF_8)));
+            LOG.error(LOG_ERR_MSG, msg, e);
+            ChannelFuture cf = ctx.writeAndFlush(
+                    Unpooled.copiedBuffer((ERR_MSG + e.getMessage() + "\n").getBytes(StandardCharsets.UTF_8)));
             if (!cf.isSuccess()) {
-                log.error(Constants.ERR_WRITING_RESPONSE, cf.cause());
+                LOG.error(Constants.ERR_WRITING_RESPONSE, cf.cause());
             }
         }
     }
 
-    private Pair<TimelyBalancedHost,TcpClient> getClient(String metric, boolean metricRequest) {
+    private Pair<TimelyBalancedHost, TcpClient> getClient(String metric, boolean metricRequest) {
         TcpClient client = null;
         TimelyBalancedHost k = null;
         int failures = 0;
         while (client == null) {
             try {
-                k = (metricRequest == true) ? metricResolver.getHostPortKeyIngest(metric) : metricResolver.getHostPortKey(metric);
+                k = (metricRequest == true) ? metricResolver.getHostPortKeyIngest(metric)
+                        : metricResolver.getHostPortKey(metric);
                 client = tcpClientPool().borrowObject(k);
             } catch (Exception e1) {
                 failures++;
                 client = null;
                 if (failures % 10 == 0) {
-                    log.error(e1.getMessage(), e1);
+                    LOG.error(e1.getMessage(), e1);
                 }
                 try {
                     Thread.sleep(failures < 10 ? 500 : 60000);

balancer/src/main/java/timely/balancer/netty/udp/UdpRelayHandler.java

@@ -2,25 +2,24 @@ package timely.balancer.netty.udp;
 
 import java.nio.charset.StandardCharsets;
 
-import org.apache.commons.lang3.tuple.Pair;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
+import org.apache.commons.lang3.tuple.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import timely.api.request.MetricRequest;
 import timely.api.request.UdpRequest;
-import timely.balancer.component.MetricResolver;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.balancer.connection.udp.UdpClientPool;
+import timely.balancer.resolver.MetricResolver;
 import timely.client.udp.UdpClient;
 import timely.netty.Constants;
 
 public class UdpRelayHandler extends SimpleChannelInboundHandler<UdpRequest> {
 
-    private static final Logger log = LoggerFactory.getLogger(UdpRelayHandler.class);
+    private static final Logger LOG = LoggerFactory.getLogger(UdpRelayHandler.class);
     private static final String LOG_ERR_MSG = "Error storing put metric: {}";
     private static final String ERR_MSG = "Error storing put metric: ";
 
@@ -34,11 +33,11 @@ public class UdpRelayHandler extends SimpleChannelInboundHandler<UdpRequest> {
 
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, UdpRequest msg) throws Exception {
-        log.trace("Received {}", msg);
+        LOG.trace("Received {}", msg);
         try {
             MetricRequest metricRequest = (MetricRequest) msg;
             String metricName = metricRequest.getMetric().getName();
-            Pair<TimelyBalancedHost,UdpClient> keyClientPair = null;
+            Pair<TimelyBalancedHost, UdpClient> keyClientPair = null;
             try {
                 keyClientPair = getClient(metricName, true);
                 keyClientPair.getRight().write(metricRequest.getLine() + "\n");
@@ -50,26 +49,28 @@ public class UdpRelayHandler extends SimpleChannelInboundHandler<UdpRequest> {
             }
 
         } catch (Exception e) {
-            log.error(LOG_ERR_MSG, msg, e);
-            ChannelFuture cf = ctx.writeAndFlush(Unpooled.copiedBuffer((ERR_MSG + e.getMessage() + "\n").getBytes(StandardCharsets.UTF_8)));
+            LOG.error(LOG_ERR_MSG, msg, e);
+            ChannelFuture cf = ctx.writeAndFlush(
+                    Unpooled.copiedBuffer((ERR_MSG + e.getMessage() + "\n").getBytes(StandardCharsets.UTF_8)));
             if (!cf.isSuccess()) {
-                log.error(Constants.ERR_WRITING_RESPONSE, cf.cause());
+                LOG.error(Constants.ERR_WRITING_RESPONSE, cf.cause());
             }
         }
     }
 
-    private Pair<TimelyBalancedHost,UdpClient> getClient(String metric, boolean metricRequest) {
+    private Pair<TimelyBalancedHost, UdpClient> getClient(String metric, boolean metricRequest) {
         UdpClient client = null;
         TimelyBalancedHost k = null;
         int failures = 0;
         while (client == null) {
             try {
-                k = (metricRequest == true) ? metricResolver.getHostPortKeyIngest(metric) : metricResolver.getHostPortKey(metric);
+                k = (metricRequest == true) ? metricResolver.getHostPortKeyIngest(metric)
+                        : metricResolver.getHostPortKey(metric);
                 client = udpClientPool.borrowObject(k);
             } catch (Exception e1) {
                 failures++;
                 client = null;
-                log.error(e1.getMessage(), e1);
+                LOG.error(e1.getMessage(), e1);
                 try {
                     Thread.sleep(failures < 10 ? 500 : 60000);
                 } catch (InterruptedException e2) {

balancer/src/main/java/timely/balancer/netty/ws/WsClientHandler.java

@@ -9,25 +9,23 @@ import javax.websocket.CloseReason;
 import javax.websocket.EndpointConfig;
 import javax.websocket.Session;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.google.common.collect.HashMultimap;
 import com.google.common.collect.Multimap;
-
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import io.netty.util.concurrent.ScheduledFuture;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import timely.api.response.TimelyException;
-import timely.auth.TimelyAuthenticationToken;
 import timely.auth.util.ProxiedEntityUtils;
 import timely.client.websocket.ClientHandler;
+import timely.netty.http.auth.TimelyAuthenticationToken;
 
 public class WsClientHandler extends ClientHandler {
 
-    private static final Logger log = LoggerFactory.getLogger(WsClientHandler.class);
+    private static final Logger LOG = LoggerFactory.getLogger(WsClientHandler.class);
     private final ChannelHandlerContext ctx;
     private final TimelyAuthenticationToken token;
     private ScheduledFuture<?> ping;
@@ -36,22 +34,22 @@ public class WsClientHandler extends ClientHandler {
         this.ctx = ctx;
         this.token = token;
         this.ping = this.ctx.executor().scheduleAtFixedRate(() -> {
-            log.trace("Sending ping on channel {}", ctx.channel());
+            LOG.trace("Sending ping on channel {}", ctx.channel());
             ctx.writeAndFlush(new PingWebSocketFrame());
         }, pingRate, pingRate, TimeUnit.SECONDS);
     }
 
     @Override
-    public void beforeRequest(Map<String,List<String>> headers) {
+    public void beforeRequest(Map<String, List<String>> headers) {
         if (token.getClientCert() != null) {
-            Multimap<String,String> proxyRequestHeaders = HashMultimap.create();
+            Multimap<String, String> proxyRequestHeaders = HashMultimap.create();
             ProxiedEntityUtils.addProxyHeaders(proxyRequestHeaders, token.getClientCert());
             for (String s : proxyRequestHeaders.keySet()) {
                 headers.put(s, new ArrayList<>(proxyRequestHeaders.get(s)));
             }
         }
 
-        Multimap<String,String> originalRequestHeaders = token.getHttpHeaders();
+        Multimap<String, String> originalRequestHeaders = token.getHttpHeaders();
         for (String s : originalRequestHeaders.keySet()) {
             if (!headers.containsKey(s)) {
                 // add pre-existing values if key does not exist in proxyRequest
@@ -64,7 +62,7 @@ public class WsClientHandler extends ClientHandler {
     public void onOpen(Session session, EndpointConfig config) {
         session.addMessageHandler(String.class, message -> {
             ctx.writeAndFlush(new TextWebSocketFrame(message));
-            log.debug("Message received on Websocket session {}: {}", session.getId(), message);
+            LOG.debug("Message received on Websocket session {}: {}", session.getId(), message);
         });
     }
 
@@ -72,8 +70,9 @@ public class WsClientHandler extends ClientHandler {
     public void onClose(Session session, CloseReason reason) {
         super.onClose(session, reason);
         if (!reason.getCloseCode().equals(CloseReason.CloseCodes.NORMAL_CLOSURE)) {
-            log.error("Abnormal close: " + reason.getReasonPhrase());
-            Exception e = new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), reason.getReasonPhrase(), "");
+            LOG.error("Abnormal close: " + reason.getReasonPhrase());
+            Exception e = new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), reason.getReasonPhrase(),
+                    "");
             WsRelayHandler.sendErrorResponse(ctx, HttpResponseStatus.INTERNAL_SERVER_ERROR, e);
         }
         if (!ping.isCancelled()) {
@@ -84,7 +83,7 @@ public class WsClientHandler extends ClientHandler {
     @Override
     public void onError(Session session, Throwable t) {
         super.onError(session, t);
-        log.error(t.getMessage(), t);
+        LOG.error(t.getMessage(), t);
         Exception e = new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), t.getMessage(), "");
         WsRelayHandler.sendErrorResponse(ctx, HttpResponseStatus.INTERNAL_SERVER_ERROR, e);
         if (!ping.isCancelled()) {

balancer/src/main/java/timely/balancer/netty/ws/WsClientHolder.java

@@ -4,16 +4,15 @@ import java.io.IOException;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.balancer.connection.ws.WsClientPool;
 import timely.client.websocket.subscription.WebSocketSubscriptionClient;
 
 public class WsClientHolder {
 
-    private static final Logger log = LoggerFactory.getLogger(WsClientHolder.class);
-    private WebSocketSubscriptionClient client;
-    private TimelyBalancedHost host;
+    private static final Logger LOG = LoggerFactory.getLogger(WsClientHolder.class);
+    private WebSocketSubscriptionClient client = null;
+    private TimelyBalancedHost host = null;
 
     public WsClientHolder(TimelyBalancedHost host, WebSocketSubscriptionClient client) {
         this.host = host;
@@ -33,7 +32,7 @@ public class WsClientHolder {
             try {
                 client.close();
             } catch (IOException e) {
-                log.error("Error closing web socket client", e);
+                LOG.error("Error closing web socket client", e);
             }
             wsClientPool.returnObject(host, client);
             client = null;

balancer/src/main/java/timely/balancer/netty/ws/WsRelayHandler.java

@@ -5,15 +5,11 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
 import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
 import com.google.common.collect.Multimap;
-
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
@@ -21,30 +17,32 @@ import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
-import timely.api.request.websocket.AddSubscription;
-import timely.api.request.websocket.CloseSubscription;
-import timely.api.request.websocket.CreateSubscription;
-import timely.api.request.websocket.RemoveSubscription;
-import timely.api.request.websocket.SubscriptionRequest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import timely.api.request.subscription.AddSubscription;
+import timely.api.request.subscription.CloseSubscription;
+import timely.api.request.subscription.CreateSubscription;
+import timely.api.request.subscription.RemoveSubscription;
+import timely.api.request.subscription.SubscriptionRequest;
 import timely.api.response.TimelyException;
 import timely.auth.util.ProxiedEntityUtils;
-import timely.balancer.component.MetricResolver;
-import timely.balancer.configuration.BalancerWebsocketProperties;
+import timely.balancer.configuration.BalancerConfiguration;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.balancer.connection.ws.WsClientPool;
+import timely.balancer.resolver.MetricResolver;
 import timely.client.websocket.subscription.WebSocketSubscriptionClient;
 import timely.netty.http.TimelyHttpHandler;
 
 public class WsRelayHandler extends SimpleChannelInboundHandler<SubscriptionRequest> implements TimelyHttpHandler {
 
-    private static final Logger log = LoggerFactory.getLogger(WsRelayHandler.class);
+    private static final Logger LOG = LoggerFactory.getLogger(WsRelayHandler.class);
 
-    private BalancerWebsocketProperties balancerWebsocketProperties;
-    private WsClientPool wsClientPool;
+    private final BalancerConfiguration balancerConfig;
+    private final WsClientPool wsClientPool;
     private MetricResolver metricResolver;
     static private ObjectMapper mapper;
-    Map<String,Map<String,WsClientHolder>> wsClients = new ConcurrentHashMap<>();
-    Map<String,ChannelHandlerContext> wsSubscriptions = new ConcurrentHashMap<>();
+    Map<String, Map<String, WsClientHolder>> wsClients = new ConcurrentHashMap<>();
+    Map<String, ChannelHandlerContext> wsSubscriptions = new ConcurrentHashMap<>();
 
     static {
         mapper = new ObjectMapper();
@@ -54,8 +52,9 @@ public class WsRelayHandler extends SimpleChannelInboundHandler<SubscriptionRequ
         mapper.configure(SerializationFeature.WRAP_ROOT_VALUE, true);
     }
 
-    public WsRelayHandler(BalancerWebsocketProperties balancerWebsocketProperties, MetricResolver metricResolver, WsClientPool wsClientPool) {
-        this.balancerWebsocketProperties = balancerWebsocketProperties;
+    public WsRelayHandler(BalancerConfiguration balancerConfig, MetricResolver metricResolver,
+            WsClientPool wsClientPool) {
+        this.balancerConfig = balancerConfig;
         this.metricResolver = metricResolver;
         this.wsClientPool = wsClientPool;
     }
@@ -68,7 +67,7 @@ public class WsRelayHandler extends SimpleChannelInboundHandler<SubscriptionRequ
         try {
             String metric = null;
 
-            Multimap<String,String> headers = msg.getRequestHeaders();
+            Multimap<String, String> headers = msg.getRequestHeaders();
             ProxiedEntityUtils.addProxyHeaders(headers, msg.getToken().getClientCert());
 
             if (msg instanceof CreateSubscription) {
@@ -84,13 +83,14 @@ public class WsRelayHandler extends SimpleChannelInboundHandler<SubscriptionRequ
 
                     @Override
                     public void operationComplete(ChannelFuture future) throws Exception {
-                        Map<String,WsClientHolder> metricToClientMap = wsClients.get(currentSubscriptionId);
+                        Map<String, WsClientHolder> metricToClientMap = wsClients.get(currentSubscriptionId);
 
                         if (metricToClientMap != null) {
                             synchronized (metricToClientMap) {
                                 // close all clients for this subscriptionId
-                                log.debug("Channel closed, closing subscriptions for subscriptionId:{}", currentSubscriptionId);
-                                for (Map.Entry<String,WsClientHolder> entry : metricToClientMap.entrySet()) {
+                                LOG.info("Channel closed, closing subscriptions for subscriptionId:%s",
+                                        currentSubscriptionId);
+                                for (Map.Entry<String, WsClientHolder> entry : metricToClientMap.entrySet()) {
                                     entry.getValue().close(wsClientPool);
                                 }
                                 wsClients.remove(currentSubscriptionId);
@@ -108,10 +108,12 @@ public class WsRelayHandler extends SimpleChannelInboundHandler<SubscriptionRequ
                 if (origContext == null) {
                     // send error because user never sent a CreateSubscription
                     // and we have nowhere to send the results
-                    log.info("ChannelHandlerContext not found for subscriptionId:{} - createSubscription not called?", subscriptionId);
-                    sendErrorResponse(ctx, HttpResponseStatus.BAD_REQUEST, new IllegalArgumentException("Must call create first"));
+                    LOG.info("ChannelHandlerContext not found for subscriptionId:%s - createSubscription not called?",
+                            subscriptionId);
+                    sendErrorResponse(ctx, HttpResponseStatus.BAD_REQUEST,
+                            new IllegalArgumentException("Must call create first"));
                 }
-                Map<String,WsClientHolder> metricToClientMap = wsClients.get(subscriptionId);
+                Map<String, WsClientHolder> metricToClientMap = wsClients.get(subscriptionId);
                 if (metricToClientMap == null) {
                     metricToClientMap = new HashMap<>();
                     wsClients.put(subscriptionId, metricToClientMap);
@@ -122,13 +124,14 @@ public class WsRelayHandler extends SimpleChannelInboundHandler<SubscriptionRequ
                     if (hostClientHolder == null) {
                         k = metricResolver.getHostPortKey(metric);
                         client = wsClientPool.borrowObject(k);
-                        client.open(new WsClientHandler(origContext, add.getToken(), (balancerWebsocketProperties.getTimeout() / 2)));
+                        client.open(new WsClientHandler(origContext, add.getToken(),
+                                (balancerConfig.getWebsocket().getTimeout() / 2)));
                         metricToClientMap.put(metric, new WsClientHolder(k, client));
                     } else {
                         client = hostClientHolder.getClient();
                     }
                 }
-                Map<String,String> tags = null;
+                Map<String, String> tags = null;
 
                 Long startTime = 0L;
                 Long endTime = 0L;
@@ -152,25 +155,25 @@ public class WsRelayHandler extends SimpleChannelInboundHandler<SubscriptionRequ
                 RemoveSubscription remove = (RemoveSubscription) msg;
                 metric = remove.getMetric();
                 subscriptionId = remove.getSubscriptionId();
-                Map<String,WsClientHolder> metricToClientMap = wsClients.get(subscriptionId);
+                Map<String, WsClientHolder> metricToClientMap = wsClients.get(subscriptionId);
                 if (metricToClientMap != null) {
                     synchronized (metricToClientMap) {
                         WsClientHolder hostClientHolder = metricToClientMap.get(metric);
                         if (hostClientHolder != null) {
                             hostClientHolder.getClient().removeSubscription(metric);
                         } else {
-                            log.info("client not found for subscriptionId:{} metric:{}", subscriptionId, metric);
+                            LOG.info("client not found for subscriptionId:%s metric:%s", subscriptionId, metric);
                         }
                     }
                 }
             } else if (msg instanceof CloseSubscription) {
                 CloseSubscription close = (CloseSubscription) msg;
                 subscriptionId = close.getSubscriptionId();
-                Map<String,WsClientHolder> metricToClientMap = wsClients.get(subscriptionId);
+                Map<String, WsClientHolder> metricToClientMap = wsClients.get(subscriptionId);
                 if (metricToClientMap != null) {
                     synchronized (metricToClientMap) {
                         // close all clients for this subscriptionId
-                        for (Map.Entry<String,WsClientHolder> entry : metricToClientMap.entrySet()) {
+                        for (Map.Entry<String, WsClientHolder> entry : metricToClientMap.entrySet()) {
                             entry.getValue().close(wsClientPool);
                         }
                     }
@@ -180,11 +183,11 @@ public class WsRelayHandler extends SimpleChannelInboundHandler<SubscriptionRequ
                 ctx.writeAndFlush(new CloseWebSocketFrame(1000, "Client requested close."));
             }
         } catch (ConnectException e1) {
-            log.error(e1.getMessage(), e1);
+            LOG.error(e1.getMessage(), e1);
             ChannelHandlerContext errorCtx = origContext == null ? ctx : origContext;
             sendErrorResponse(errorCtx, HttpResponseStatus.INTERNAL_SERVER_ERROR, e1);
         } catch (Exception e2) {
-            log.error(e2.getMessage(), e2);
+            LOG.error(e2.getMessage(), e2);
             ChannelHandlerContext errorCtx = origContext == null ? ctx : origContext;
             sendErrorResponse(errorCtx, HttpResponseStatus.INTERNAL_SERVER_ERROR, e2);
         }
@@ -198,7 +201,7 @@ public class WsRelayHandler extends SimpleChannelInboundHandler<SubscriptionRequ
             String json = mapper.writeValueAsString(te);
             ctx.writeAndFlush(new TextWebSocketFrame(json));
         } catch (JsonProcessingException e) {
-            log.error("Error serializing exception");
+            LOG.error("Error serializing exception");
         }
     }
 }

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

@@ -0,0 +1,15 @@
+package timely.balancer.resolver;
+
+import timely.balancer.connection.TimelyBalancedHost;
+
+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/resources/config/bootstrap.yml

@@ -1,6 +0,0 @@
-spring:
-  application:
-    name: timely-balancer
-  config:
-    name: timely-balancer
-    location: ${CONF_DIR}/

balancer/src/main/spotbugs/excludes.xml

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <FindBugsFilter>
   <Match>
-    <Class name="timely.balancer.component.Balancer" />
+    <Class name="timely.balancer.Balancer" />
     <Method name="shutdown" />
     <Bug pattern="UC_USELESS_OBJECT" />
   </Match>
@@ -11,11 +11,8 @@
     <Bug pattern="EI_EXPOSE_REP" />
   </Match>
   <Match>
-    <!-- these are probably avoidable, but needs further triage to fix -->
-    <Bug pattern="EI_EXPOSE_REP" />
-  </Match>
-  <Match>
-    <!-- these are probably avoidable, but needs further triage to fix -->
-    <Bug pattern="EI_EXPOSE_REP2" />
+    <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/component/TestBalancer.java

@@ -1,33 +0,0 @@
-package timely.balancer.component;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.context.ApplicationContext;
-import org.springframework.context.annotation.Primary;
-import org.springframework.stereotype.Component;
-
-import timely.balancer.configuration.BalancerHttpProperties;
-import timely.balancer.configuration.BalancerProperties;
-import timely.balancer.configuration.BalancerServerProperties;
-import timely.balancer.configuration.BalancerWebsocketProperties;
-import timely.common.component.AuthenticationService;
-import timely.common.configuration.SecurityProperties;
-import timely.common.configuration.SslClientProperties;
-import timely.common.configuration.SslServerProperties;
-import timely.common.configuration.ZookeeperProperties;
-
-@Primary
-@Component
-public class TestBalancer extends Balancer {
-    private static final Logger log = LoggerFactory.getLogger(TestBalancer.class);
-
-    public TestBalancer(ApplicationContext applicationContext, AuthenticationService authenticationService, MetricResolver metricResolver,
-                    BalancerProperties balancerProperties, ZookeeperProperties zookeeperProperties, SecurityProperties securityProperties,
-                    SslServerProperties sslServerProperties, SslClientProperties sslClientProperties, BalancerServerProperties balancerServerProperties,
-                    BalancerHttpProperties balancerHttpProperties, BalancerWebsocketProperties balancerWebsocketProperties) {
-        super(applicationContext, authenticationService, metricResolver, balancerProperties, zookeeperProperties, securityProperties, sslServerProperties,
-                        sslClientProperties, balancerServerProperties, balancerHttpProperties, balancerWebsocketProperties);
-        DEFAULT_EVENT_LOOP_THREADS = 1;
-        log.info("Starting TestBalancer");
-    }
-}

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

@@ -1,33 +0,0 @@
-package timely.balancer.component;
-
-import java.util.List;
-
-import org.springframework.context.annotation.Primary;
-import org.springframework.stereotype.Component;
-
-import timely.balancer.configuration.BalancerProperties;
-import timely.balancer.configuration.BalancerServerProperties;
-import timely.balancer.connection.TimelyBalancedHost;
-
-@Primary
-@Component
-public class TestHealthChecker extends HealthChecker {
-
-    public TestHealthChecker(BalancerProperties balancerProperties, BalancerServerProperties balancerServerProperties) {
-        super(balancerProperties, balancerServerProperties);
-    }
-
-    @Override
-    public void setup() {
-
-    }
-
-    @Override
-    public void close() {
-
-    }
-
-    public List<TimelyBalancedHost> getTimelyHosts() {
-        return this.timelyHosts;
-    }
-}

balancer/src/test/java/timely/balancer/component/TestMetricResolver.java

@@ -1,46 +0,0 @@
-package timely.balancer.component;
-
-import java.util.List;
-
-import org.apache.curator.framework.CuratorFramework;
-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 org.springframework.context.annotation.Primary;
-import org.springframework.stereotype.Component;
-
-import timely.balancer.configuration.BalancerProperties;
-import timely.balancer.connection.TimelyBalancedHost;
-import timely.common.configuration.CacheProperties;
-
-@Primary
-@Component
-public class TestMetricResolver extends MetricResolver {
-
-    private static final Logger log = LoggerFactory.getLogger(TestMetricResolver.class);
-
-    public TestMetricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
-                    HealthChecker healthChecker) {
-        super(curatorFramework, balancerProperties, cacheProperties, healthChecker);
-    }
-
-    protected void createAssignmentFile() {
-        try {
-            log.info("Creating assignment file: " + balancerProperties.getAssignmentFile());
-            Configuration configuration = new Configuration();
-            FileSystem fs = FileSystem.get(configuration);
-            Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
-            fs.setWriteChecksum(false);
-            fs.create(assignmentFile);
-            fs.deleteOnExit(assignmentFile);
-        } catch (Exception e) {
-            log.error(e.getMessage(), e);
-        }
-    }
-
-    public List<TimelyBalancedHost> getServerList() {
-        return this.serverList;
-    }
-}

balancer/src/test/java/timely/balancer/component/TestTimelyAuthenticationManager.java

@@ -1,20 +0,0 @@
-package timely.balancer.component;
-
-import java.util.Collections;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.context.annotation.Primary;
-import org.springframework.security.authentication.ProviderManager;
-import org.springframework.security.web.authentication.preauth.PreAuthenticatedAuthenticationProvider;
-import org.springframework.stereotype.Component;
-
-@Primary
-@Component
-public class TestTimelyAuthenticationManager extends ProviderManager {
-
-    @Autowired
-    public TestTimelyAuthenticationManager(@Qualifier("file") PreAuthenticatedAuthenticationProvider file) {
-        super(Collections.singletonList(file));
-    }
-}

balancer/src/test/java/timely/balancer/configuration/TestConfiguration.java

@@ -1,21 +0,0 @@
-package timely.balancer.configuration;
-
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-import timely.common.configuration.AccumuloProperties;
-
-@Configuration
-public class TestConfiguration {
-
-    @Bean
-    @ConditionalOnMissingBean
-    public AccumuloProperties accumuloProperties() {
-        AccumuloProperties accumuloProperties = new AccumuloProperties();
-        accumuloProperties.setInstanceName("default");
-        accumuloProperties.setUsername("root");
-        accumuloProperties.setPassword("secret");
-        return accumuloProperties;
-    }
-}

balancer/src/test/java/timely/balancer/configuration/TestCuratorConfiguration.java

@@ -1,49 +0,0 @@
-package timely.balancer.configuration;
-
-import static org.apache.accumulo.core.conf.ConfigurationTypeHelper.getTimeInMillis;
-
-import org.apache.curator.RetryPolicy;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.CuratorFrameworkFactory;
-import org.apache.curator.retry.RetryForever;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.DependsOn;
-
-import timely.common.configuration.ZookeeperProperties;
-
-/*
- * The purpose of this class is to wait for the creation of either ZooKeeperServerMain (in-memory)
- * or MiniAccumuloCluster (mini accumulo) before creating a CuratorFramework so that the components
- * that need CuratorFramework autowired will wait.
- */
-@Configuration
-public class TestCuratorConfiguration {
-
-    private static final Logger log = LoggerFactory.getLogger(TestCuratorConfiguration.class);
-
-    @Bean(destroyMethod = "close")
-    @ConditionalOnProperty(name = "timely.in-memory-accumulo.enabled", havingValue = "true")
-    @DependsOn("ZooKeeperServerMain")
-    public CuratorFramework inMemoryCuratorFramework(ZookeeperProperties zookeeperProperties) {
-        RetryPolicy retryPolicy = new RetryForever(1000);
-        int timeout = Long.valueOf(getTimeInMillis(zookeeperProperties.getTimeout())).intValue();
-        CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(zookeeperProperties.getServers(), timeout, 10000, retryPolicy);
-        curatorFramework.start();
-        return curatorFramework;
-    }
-
-    @Bean(destroyMethod = "close")
-    @ConditionalOnProperty(name = "timely.mini-accumulo.enabled", havingValue = "true")
-    @DependsOn("MiniAccumuloCluster")
-    public CuratorFramework miniAccumuloCuratorFramework(ZookeeperProperties zookeeperProperties) {
-        RetryPolicy retryPolicy = new RetryForever(1000);
-        int timeout = Long.valueOf(getTimeInMillis(zookeeperProperties.getTimeout())).intValue();
-        CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(zookeeperProperties.getServers(), timeout, 10000, retryPolicy);
-        curatorFramework.start();
-        return curatorFramework;
-    }
-}

balancer/src/test/java/timely/balancer/configuration/TestWatcherConfiguration.java

@@ -1,23 +0,0 @@
-package timely.balancer.configuration;
-
-import org.apache.accumulo.core.client.AccumuloClient;
-import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Scope;
-
-import timely.balancer.test.BalancerTestRule;
-import timely.common.configuration.MiniAccumuloProperties;
-import timely.common.configuration.TimelyProperties;
-
-@Configuration
-@EnableConfigurationProperties({MiniAccumuloProperties.class, TimelyProperties.class})
-public class TestWatcherConfiguration {
-
-    @Bean
-    @Scope("prototype")
-    public BalancerTestRule timelyTestWatcherAccumulo(AccumuloClient accumuloClient, TimelyProperties timelyProperties) throws Exception {
-
-        return new BalancerTestRule(timelyProperties, accumuloClient);
-    }
-}

balancer/src/test/java/timely/balancer/configuration/TestZookeeperConfiguration.java

@@ -1,80 +0,0 @@
-package timely.balancer.configuration;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.ServerSocket;
-import java.nio.file.Files;
-import java.util.Properties;
-import java.util.concurrent.Executors;
-
-import org.apache.zookeeper.server.ServerConfig;
-import org.apache.zookeeper.server.ZooKeeperServerMain;
-import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-import timely.common.configuration.ZookeeperProperties;
-
-@Configuration
-@ConditionalOnProperty(name = "timely.in-memory-accumulo.enabled", havingValue = "true")
-public class TestZookeeperConfiguration {
-
-    private static final Logger log = LoggerFactory.getLogger(TestZookeeperConfiguration.class);
-    private static final int MAX_PORT = 100;
-    private int port;
-
-    private ZookeeperProperties zookeeperProperties;
-
-    @Bean(name = "ZooKeeperServerMain", destroyMethod = "close")
-    public ZooKeeperServerMain zooKeeperServerMain(ZookeeperProperties zookeeperProperties) throws Exception {
-        this.zookeeperProperties = zookeeperProperties;
-        this.port = getNextPort(2181);
-        this.zookeeperProperties.setServers(String.format("localhost:%d", port));
-
-        File tempDir;
-        try {
-            tempDir = Files.createTempDirectory("zookeeper_temp").toFile();
-            tempDir.deleteOnExit();
-        } catch (IOException e) {
-            throw new RuntimeException("Unable to create temp directory for mini accumulo cluster");
-        }
-
-        Properties properties = new Properties();
-        properties.put("dataDir", tempDir.getAbsolutePath() + "/data");
-        properties.put("clientPort", port);
-        properties.put("skipACL", "true");
-        properties.put("leaderServes", "yes");
-        properties.put("4lw.commands.whitelist", "*");
-        properties.put("admin.enableServer", "false");
-        QuorumPeerConfig quorumPeerConfig = new QuorumPeerConfig();
-        quorumPeerConfig.parseProperties(properties);
-
-        ZooKeeperServerMain zookeeper = new ZooKeeperServerMain();
-        final ServerConfig config = new ServerConfig();
-        config.readFrom(quorumPeerConfig);
-        Executors.newSingleThreadExecutor().submit(() -> {
-            try {
-                zookeeper.runFromConfig(config);
-            } catch (Exception e) {
-                log.error(e.getMessage(), e);
-            }
-        });
-        log.info("Started TestZookeeper on port {}", port);
-        return zookeeper;
-    }
-
-    private int getNextPort(int start) {
-        for (int port = start; port < start + MAX_PORT; ++port) {
-            try {
-                new ServerSocket(port).close();
-                return port;
-            } catch (IOException e) {
-
-            }
-        }
-        return -1;
-    }
-}

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/java/timely/balancer/test/BalancerTestRule.java

@@ -1,61 +0,0 @@
-package timely.balancer.test;
-
-import org.apache.accumulo.core.client.AccumuloClient;
-import org.junit.rules.TestWatcher;
-import org.junit.runner.Description;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import timely.common.configuration.TimelyProperties;
-import timely.util.TestUtils;
-
-public class BalancerTestRule extends TestWatcher {
-
-    private static final Logger log = LoggerFactory.getLogger(BalancerTestRule.class);
-    private TimelyProperties timelyProperties;
-    private AccumuloClient accumuloClient;
-
-    public BalancerTestRule(TimelyProperties timelyProperties, AccumuloClient accumuloClient) {
-        this.timelyProperties = timelyProperties;
-        this.accumuloClient = accumuloClient;
-    }
-
-    @Override
-    protected void failed(Throwable e, Description description) {
-        log.error("Test {}:{} failed", description.getTestClass().getCanonicalName(), description.getMethodName());
-        log.error(e.getMessage(), e);
-        if (accumuloClient != null) {
-            if (accumuloClient.tableOperations().exists(timelyProperties.getMetricsTable())) {
-                separator();
-                log.info("Contents of {}", timelyProperties.getMetricsTable());
-                separator();
-                TestUtils.printTable(accumuloClient, timelyProperties.getMetricsTable(), true);
-                separator();
-            }
-            if (accumuloClient.tableOperations().exists(timelyProperties.getMetaTable())) {
-                log.info("Contents of {}", timelyProperties.getMetaTable());
-                separator();
-                TestUtils.printTable(accumuloClient, timelyProperties.getMetaTable(), false);
-                separator();
-            }
-        }
-    }
-
-    private void separator() {
-        log.info("-----------------------------------------------------------------------------------");
-    }
-
-    @Override
-    protected void starting(Description description) {
-        separator();
-        log.info("Starting test {}:{}", description.getTestClass().getCanonicalName(), description.getMethodName());
-        separator();
-    }
-
-    @Override
-    protected void finished(Description description) {
-        separator();
-        log.info("Finished test {}:{}", description.getTestClass().getCanonicalName(), description.getMethodName());
-        separator();
-    }
-}

balancer/src/test/java/timely/balancer/test/MetricResolverIT.java

@@ -1,137 +0,0 @@
-package timely.balancer.test;
-
-import static timely.Constants.SERVICE_DISCOVERY_PATH;
-
-import java.util.List;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.state.ConnectionState;
-import org.apache.curator.x.discovery.ServiceCache;
-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.curator.x.discovery.details.ServiceCacheListener;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.junit.runner.RunWith;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.test.context.junit4.SpringRunner;
-
-import timely.ServerDetails;
-import timely.balancer.component.TestHealthChecker;
-import timely.balancer.component.TestMetricResolver;
-import timely.balancer.connection.TimelyBalancedHost;
-import timely.test.IntegrationTest;
-
-@Category(IntegrationTest.class)
-@RunWith(SpringRunner.class)
-@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
-public class MetricResolverIT {
-
-    private static final Logger log = LoggerFactory.getLogger(MetricResolverIT.class);
-
-    @Autowired
-    @Rule
-    public BalancerTestRule balancerTestRule;
-
-    @Autowired
-    private TestMetricResolver metricResolver;
-
-    @Autowired
-    private TestHealthChecker healthChecker;
-
-    @Autowired
-    private CuratorFramework curatorFramework;
-
-    private ServiceDiscovery<ServerDetails> discovery;
-    private ServiceCache<ServerDetails> serviceCache;
-
-    @Before
-    public void setup() throws Exception {
-        discovery = ServiceDiscoveryBuilder.builder(ServerDetails.class).client(curatorFramework).basePath(SERVICE_DISCOVERY_PATH).build();
-        discovery.start();
-        serviceCache = discovery.serviceCacheBuilder().name("timely-server").build();
-    }
-
-    @Test
-    public void ServiceRegistrationTest() {
-        LinkedBlockingQueue<List<ServiceInstance<ServerDetails>>> updateQueue = getServiceUpdateQueue(serviceCache);
-        registerService(curatorFramework, TimelyBalancedHost.of("host1", 5242, 5243, 5244, 5245, null));
-        registerService(curatorFramework, TimelyBalancedHost.of("host1", 5342, 5343, 5344, 5345, null));
-        registerService(curatorFramework, TimelyBalancedHost.of("host1", 5442, 5443, 5444, 5445, null));
-        registerService(curatorFramework, TimelyBalancedHost.of("host1", 5542, 5543, 5544, 5545, null));
-        try {
-            List<ServiceInstance<ServerDetails>> currentUpdate;
-            List<ServiceInstance<ServerDetails>> lastUpdate = null;
-            do {
-                currentUpdate = updateQueue.poll(200, TimeUnit.MILLISECONDS);
-                if (currentUpdate != null) {
-                    lastUpdate = currentUpdate;
-                }
-            } while (currentUpdate != null);
-            Assert.assertNotNull(lastUpdate);
-            Assert.assertEquals(4, lastUpdate.size());
-        } catch (InterruptedException e) {
-            Assert.fail("" + e.getMessage());
-        }
-        Assert.assertEquals(4, metricResolver.getServerList().size());
-        Assert.assertEquals(4, healthChecker.getTimelyHosts().size());
-    }
-
-    public void registerService(CuratorFramework curatorFramework, TimelyBalancedHost server) {
-        try {
-            ServerDetails payload = new ServerDetails();
-            payload.setHost(server.getHost());
-            payload.setTcpPort(server.getTcpPort());
-            payload.setHttpPort(server.getHttpPort());
-            payload.setWsPort(server.getWsPort());
-            payload.setUdpPort(server.getUdpPort());
-
-            ServiceInstanceBuilder<ServerDetails> builder = ServiceInstance.builder();
-            String serviceName = server.getHost() + ":" + server.getTcpPort();
-            ServiceInstance<ServerDetails> serviceInstance = builder.id(serviceName).name("timely-server").address(server.getHost()).port(server.getTcpPort())
-                            .payload(payload).build();
-
-            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());
-        }
-    }
-
-    private LinkedBlockingQueue<List<ServiceInstance<ServerDetails>>> getServiceUpdateQueue(ServiceCache<ServerDetails> serviceCache) {
-
-        LinkedBlockingQueue<List<ServiceInstance<ServerDetails>>> updates = new LinkedBlockingQueue<>();
-        try {
-
-            ServiceCacheListener listener = new ServiceCacheListener() {
-
-                @Override
-                public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
-
-                }
-
-                @Override
-                public void cacheChanged() {
-                    updates.add(serviceCache.getInstances());
-                }
-            };
-            serviceCache.addListener(listener);
-            serviceCache.start();
-        } catch (Exception e) {
-            log.error(e.getMessage(), e);
-        }
-        return updates;
-    }
-}

balancer/src/test/java/timely/balancer/test/TestServerInfo.java

@@ -1,48 +0,0 @@
-package timely.balancer.test;
-
-public class TestServerInfo {
-
-    public String host;
-    public String ip;
-    public Integer tcpPort;
-    public Integer httpPort;
-    public Integer wsPort;
-    public Integer udpPort;
-
-    public TestServerInfo(String host, String ip, Integer tcpPort, Integer httpPort, Integer wsPort, Integer udpPort) {
-        this.host = host;
-        this.ip = ip;
-        this.tcpPort = tcpPort;
-        this.httpPort = httpPort;
-        this.wsPort = wsPort;
-        this.udpPort = udpPort;
-    }
-
-    static public TestServerInfo of(String host, String ip, Integer tcpPort, Integer httpPort, Integer wsPort, Integer udpPort) {
-        return new TestServerInfo(host, ip, tcpPort, httpPort, wsPort, udpPort);
-    }
-
-    public String getHost() {
-        return host;
-    }
-
-    public String getIp() {
-        return ip;
-    }
-
-    public Integer getTcpPort() {
-        return tcpPort;
-    }
-
-    public Integer getHttpPort() {
-        return httpPort;
-    }
-
-    public Integer getWsPort() {
-        return wsPort;
-    }
-
-    public Integer getUdpPort() {
-        return udpPort;
-    }
-}

balancer/src/test/resources/config/application.yml

@@ -1,141 +0,0 @@
-logging:
-  level:
-    ROOT: info
-    timely: debug
-    org.apache.zookeeper: error
-
-timely:
-  test: true
-  mini-accumulo:
-    enabled: true
-    rootAuths: A,B,C,D,E,F,G,H,I
-  zookeeper:
-    servers: localhost:0
-    timeout: 30s
-  security:
-    session-max-age: 86400
-    cache-expiration-minutes: 60
-    cache-refresh-minutes: 5
-    allow-anonymous-http-access: true
-    allow-anonymous-ws-access: true
-    authorizedUsers:
-      - subjectDn: cn=example.com
-        roles: G,H,I
-        auths: D,E,F
-        userType: USER
-      - subjectDn: cn=test a. user, ou=example developers, o=example corp, c=us
-        issuerDn: cn=example corp ca, o=example corp, c=us
-        roles: G,H,I
-        auths: A,B,C
-        userType: USER
-    serverSsl:
-      useGeneratedKeypair: true
-      ignoreSslHandshakeErrors: true
-      useOpenssl: true
-  cache:
-    enabled: false
-    metric-age-off-hours:
-      default: 26
-    nonCachedMetrics:
-  balancer:
-    checkServerHealthInterval: 10000
-    serverFailuresBeforeDown: 6
-    serverSuccessesBeforeUp: 1
-    assignmentFile: file://${java.io.tmpdir}/timely/balancer/assignments.csv
-    server:
-      ip: 127.0.0.1
-      tcpPort: 4242
-      numTcpPools: 1
-      tcpBufferSize: 100
-      tcpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-      udpPort: 4245
-      udpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-    http:
-      host: localhost
-      ip: 127.0.0.1
-      port: 4243
-      redirect-path: /secure-me
-      strict-transport-max-age: 604800
-      cors:
-        allow-any-origin: false
-        allow-null-origin: false
-        allow-credentials: true
-        allowed-origins:
-        allowed-methods: DELETE,GET,HEAD,OPTIONS,PUT,POST
-        allowed-headers: content-type
-      httpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-    websocket:
-      ip: 127.0.0.1
-      port: 5244
-      timeout: 60
-      subscription-lag: 120
-      scanner-batch-size: 10000
-      flush-interval-seconds: 30
-      scanner-read-ahead: 1
-      subscription-batch-size: 5000
-      incomingBufferSize: 50000000
-      wsClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false

balancer/src/test/resources/config/bootstrap.yml

@@ -1,3 +0,0 @@
-spring:
-  application:
-    name: timely-balancer

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>

balancer/src/test/resources/log4j2.yml

@@ -1,24 +0,0 @@
-Configuration:
-  status: warn
-  monitorInterval: 60
-
-  Properties:
-    Property:
-      - name: PID
-        value: "????"
-      - name: LOG_PATTERN
-        value: "%clr{%d{yyyy-MM-dd HH:mm:ss.SSS}}{faint} %clr{%5p} %clr{${sys:PID}}{magenta} %clr{---}{faint} %clr{[%15.15t]}{faint} %clr{%-40.40c{1.}}{cyan} %clr{:}{faint} %m%n%wEx"
-
-  Appenders:
-    Console:
-      name: Console
-      target: SYSTEM_OUT
-      follow: true
-      PatternLayout:
-        pattern: "${LOG_PATTERN}"
-
-  Loggers:
-    Root:
-      level: warn
-      AppenderRef:
-        - ref: Console

client/pom.xml

@@ -4,13 +4,13 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>1.2.8-SNAPSHOT</version>
   </parent>
   <artifactId>timely-client</artifactId>
   <properties>
     <!-- PEP 440 compliant version gets substituted into src/main/python/template/setup.py -->
     <!-- use .dev0 instead of -SNAPSHOT -->
-    <python.client.version>2.1.0.dev0</python.client.version>
+    <python.client.version>1.2.8.dev0</python.client.version>
   </properties>
   <dependencies>
     <dependency>
@@ -53,10 +53,6 @@
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>
     </dependency>
-    <dependency>
-      <groupId>timely</groupId>
-      <artifactId>timely-common</artifactId>
-    </dependency>
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-api</artifactId>
@@ -67,18 +63,21 @@
       <artifactId>log4j-core</artifactId>
       <scope>runtime</scope>
     </dependency>
-    <!--
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-slf4j-impl</artifactId>
       <scope>runtime</scope>
     </dependency>
-    -->
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>timely</groupId>
+      <artifactId>timely-common</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
   <build>
     <plugins>

client/src/main/java/timely/api/AddSubscription.java

@@ -7,7 +7,7 @@ public class AddSubscription extends SubscriptionRequest {
 
     private String operation = "add";
     private String metric = null;
-    private Optional<Map<String,String>> tags = Optional.empty();
+    private Optional<Map<String, String>> tags = Optional.empty();
     private Optional<Long> startTime = Optional.empty();
     private Optional<Long> endTime = Optional.empty();
     private Optional<Long> delayTime = Optional.empty();
@@ -20,7 +20,7 @@ public class AddSubscription extends SubscriptionRequest {
         return metric;
     }
 
-    public Optional<Map<String,String>> getTags() {
+    public Optional<Map<String, String>> getTags() {
         return tags;
     }
 
@@ -36,7 +36,7 @@ public class AddSubscription extends SubscriptionRequest {
         this.metric = metric;
     }
 
-    public void setTags(Optional<Map<String,String>> tags) {
+    public void setTags(Optional<Map<String, String>> tags) {
         this.tags = tags;
     }
 
@@ -71,8 +71,9 @@ public class AddSubscription extends SubscriptionRequest {
         }
         if (obj instanceof AddSubscription) {
             AddSubscription other = (AddSubscription) obj;
-            return (this.subscriptionId.equals(other.subscriptionId) && this.metric.equals(other.metric) && this.tags.equals(other.tags)
-                            && this.startTime.equals(other.startTime) && this.endTime.equals(other.endTime) && this.delayTime.equals(other.delayTime));
+            return (this.subscriptionId.equals(other.subscriptionId) && this.metric.equals(other.metric)
+                    && this.tags.equals(other.tags) && this.startTime.equals(other.startTime)
+                    && this.endTime.equals(other.endTime) && this.delayTime.equals(other.delayTime));
         } else {
             return false;
         }

client/src/main/java/timely/api/BasicAuthLogin.java

@@ -0,0 +1,45 @@
+package timely.api;
+
+public class BasicAuthLogin {
+
+    protected String username;
+    protected String password;
+
+    public String getUsername() {
+        return username;
+    }
+
+    public void setUsername(String user) {
+        this.username = user;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
+    @Override
+    public int hashCode() {
+        return (username + password).hashCode();
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (null == obj) {
+            return false;
+        }
+        if (obj == this) {
+            return true;
+        }
+        if (obj instanceof BasicAuthLogin) {
+            BasicAuthLogin other = (BasicAuthLogin) obj;
+            return (this.username.equals(other.username) && this.password.equals(other.password));
+        } else {
+            return false;
+        }
+    }
+
+}

client/src/main/java/timely/api/VersionRequest.java

@@ -1,10 +0,0 @@
-package timely.api;
-
-public class VersionRequest {
-
-    private String operation = "version";
-
-    public String getOperation() {
-        return operation;
-    }
-}

client/src/main/java/timely/api/response/MetricResponse.java

@@ -5,13 +5,9 @@ import java.util.List;
 import org.apache.commons.lang3.builder.EqualsBuilder;
 import org.apache.commons.lang3.builder.HashCodeBuilder;
 import org.apache.commons.lang3.builder.ToStringBuilder;
-
-import com.fasterxml.jackson.annotation.JsonPropertyOrder;
-
 import timely.model.Metric;
 import timely.model.Tag;
 
-@JsonPropertyOrder({"metric", "timestamp", "value", "tags", "subscriptionId", "complete"})
 public class MetricResponse {
 
     private String metric;

client/src/main/java/timely/client/http/HttpClient.java

@@ -15,8 +15,8 @@ import org.glassfish.tyrus.client.SslContextConfigurator;
 
 public class HttpClient {
 
-    public static SSLContext getSSLContext(String trustStoreFile, String trustStoreType, String trustStorePass, String keyStoreFile, String keyStoreType,
-                    String keyStorePass) {
+    public static SSLContext getSSLContext(String trustStoreFile, String trustStoreType, String trustStorePass,
+            String keyStoreFile, String keyStoreType, String keyStorePass) {
         SslContextConfigurator scc = new SslContextConfigurator();
         scc.setTrustStoreFile(trustStoreFile);
         scc.setTrustStoreType(trustStoreType);
@@ -31,10 +31,12 @@ public class HttpClient {
         return scc.createSSLContext();
     }
 
-    public static CloseableHttpClient get(String trustStoreFile, String trustStoreType, String trustStorePass, String keyStoreFile, String keyStoreType,
-                    String keyStorePass, CookieStore cookieStore, boolean hostVerificationEnabled) {
+    public static CloseableHttpClient get(String trustStoreFile, String trustStoreType, String trustStorePass,
+            String keyStoreFile, String keyStoreType, String keyStorePass, CookieStore cookieStore,
+            boolean hostVerificationEnabled) {
 
-        SSLContext ssl = getSSLContext(trustStoreFile, trustStoreType, trustStorePass, keyStoreFile, keyStoreType, keyStorePass);
+        SSLContext ssl = getSSLContext(trustStoreFile, trustStoreType, trustStorePass, keyStoreFile, keyStoreType,
+                keyStorePass);
         return get(ssl, cookieStore, hostVerificationEnabled);
     }
 
@@ -42,7 +44,8 @@ public class HttpClient {
         return get(ssl, cookieStore, hostVerificationEnabled, false);
     }
 
-    public static CloseableHttpClient get(SSLContext ssl, CookieStore cookieStore, boolean hostVerificationEnabled, boolean clientAuth) {
+    public static CloseableHttpClient get(SSLContext ssl, CookieStore cookieStore, boolean hostVerificationEnabled,
+            boolean clientAuth) {
         HostnameVerifier hostnameVerifier;
         if (hostVerificationEnabled) {
             hostnameVerifier = SSLConnectionSocketFactory.getDefaultHostnameVerifier();
@@ -52,15 +55,16 @@ public class HttpClient {
         RequestConfig defaultRequestConfig = RequestConfig.custom().setCookieSpec(CookieSpecs.STANDARD).build();
         HttpClientBuilder builder;
         if (cookieStore == null) {
-            builder = HttpClients.custom().setSSLContext(ssl).disableCookieManagement().setDefaultRequestConfig(defaultRequestConfig)
-                            .setSSLHostnameVerifier(hostnameVerifier);
+            builder = HttpClients.custom().setSSLContext(ssl).disableCookieManagement()
+                    .setDefaultRequestConfig(defaultRequestConfig).setSSLHostnameVerifier(hostnameVerifier);
         } else {
-            builder = HttpClients.custom().setSSLContext(ssl).setDefaultCookieStore(cookieStore).setDefaultRequestConfig(defaultRequestConfig)
-                            .setSSLHostnameVerifier(hostnameVerifier);
+            builder = HttpClients.custom().setSSLContext(ssl).setDefaultCookieStore(cookieStore)
+                    .setDefaultRequestConfig(defaultRequestConfig).setSSLHostnameVerifier(hostnameVerifier);
         }
 
         if (clientAuth) {
-            SSLConnectionSocketFactory sslCSF = new SSLConnectionSocketFactory(ssl, new String[] {"TLSv1.1", "TLSv1.2"}, null, hostnameVerifier);
+            SSLConnectionSocketFactory sslCSF = new SSLConnectionSocketFactory(ssl,
+                    new String[] { "TLSv1.1", "TLSv1.2" }, null, hostnameVerifier);
             builder.setSSLSocketFactory(sslCSF);
         }
         return builder.build();

client/src/main/java/timely/client/tcp/TcpClient.java

@@ -4,14 +4,13 @@ import java.io.BufferedWriter;
 import java.io.IOException;
 import java.io.OutputStreamWriter;
 import java.net.Socket;
-import java.nio.charset.StandardCharsets;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class TcpClient implements AutoCloseable {
 
-    private static final Logger log = LoggerFactory.getLogger(TcpClient.class);
+    private static final Logger LOG = LoggerFactory.getLogger(TcpClient.class);
 
     private final String host;
     private final int port;
@@ -34,7 +33,7 @@ public class TcpClient implements AutoCloseable {
 
     /**
      * Opens a TCP connection to the specified host and port
-     *
+     * 
      * @throws IOException
      *             if an error occurs
      */
@@ -46,7 +45,7 @@ public class TcpClient implements AutoCloseable {
 
     /**
      * Write a metric to Timely
-     *
+     * 
      * @param metric
      *            newline terminated string representation of Timely metric
      */
@@ -70,13 +69,13 @@ public class TcpClient implements AutoCloseable {
 
     /**
      * Closes the tcp connection to Timely
-     *
+     * 
      * @throws IOException
      *             if an error occurs
      */
     @Override
     public void close() throws IOException {
-        log.trace("Shutting down connection to Timely at {}:{}", host, port);
+        LOG.trace("Shutting down connection to Timely at {}:{}", host, port);
         if (null != sock) {
             try {
                 if (null != out) {
@@ -87,7 +86,7 @@ public class TcpClient implements AutoCloseable {
                 }
                 sock.close();
             } catch (IOException e) {
-                log.error("Error closing connection to Timely at " + host + ":" + port + ". Error: " + e.getMessage());
+                LOG.error("Error closing connection to Timely at " + host + ":" + port + ". Error: " + e.getMessage());
             } finally {
                 sock = null;
             }
@@ -97,47 +96,22 @@ public class TcpClient implements AutoCloseable {
     private synchronized int connect() {
         if (null == sock || !sock.isConnected()) {
             if (System.currentTimeMillis() > (connectTime + backoff)) {
-                OutputStreamWriter osw = null;
                 try {
                     connectTime = System.currentTimeMillis();
                     sock = new Socket(host, port);
-                    osw = new OutputStreamWriter(sock.getOutputStream(), StandardCharsets.UTF_8);
-                    out = new BufferedWriter(osw);
+                    out = new BufferedWriter(new OutputStreamWriter(sock.getOutputStream()));
                     backoff = 2000;
-                    log.trace("Connected to Timely at {}:{}", host, port);
-                } catch (Exception e) {
-                    log.error("Error connecting to Timely at {}:{} - {}", host, port, e.getMessage());
-                    if (sock != null) {
-                        try {
-                            sock.close();
-                        } catch (IOException e1) {
-                            log.error(e1.getMessage());
-                        } finally {
-                            sock = null;
-                        }
-                    }
-                    if (osw != null) {
-                        try {
-                            osw.close();
-                        } catch (IOException e1) {
-                            log.error(e1.getMessage());
-                        }
-                    }
-                    if (out != null) {
-                        try {
-                            out.close();
-                        } catch (IOException e1) {
-                            log.error(e1.getMessage());
-                        } finally {
-                            out = null;
-                        }
-                    }
+                    LOG.trace("Connected to Timely at {}:{}", host, port);
+                } catch (IOException e) {
+                    LOG.error("Error connecting to Timely at {}:" + host + ":" + port + ". Error: " + e.getMessage());
                     backoff = backoff * 2;
-                    log.info("Will retry connection in {} ms.", backoff);
+                    sock = null;
+                    out = null;
+                    LOG.info("Will retry connection in {} ms.", backoff);
                     return -1;
                 }
             } else {
-                log.warn("Not writing to Timely, waiting to reconnect");
+                LOG.warn("Not writing to Timely, waiting to reconnect");
                 return -1;
             }
         }

client/src/main/java/timely/client/udp/UdpClient.java

@@ -12,7 +12,7 @@ import org.slf4j.LoggerFactory;
 
 public class UdpClient implements AutoCloseable {
 
-    private static final Logger log = LoggerFactory.getLogger(UdpClient.class);
+    private static final Logger LOG = LoggerFactory.getLogger(UdpClient.class);
 
     private final String host;
     private final int port;
@@ -38,11 +38,12 @@ public class UdpClient implements AutoCloseable {
             throw new IllegalStateException("Must call open first");
         }
         this.packet.setData(metric.getBytes(UTF_8));
-        log.info("writing '" + metric + "' to " + this.host + ":" + this.port);
+        LOG.info("writing '" + metric + "' to " + this.host + ":" + this.port);
         this.sock.send(packet);
     }
 
-    public void flush() throws IOException {}
+    public void flush() throws IOException {
+    }
 
     public void close() throws IOException {
         if (null != sock) {

client/src/main/java/timely/client/websocket/ClientHandler.java

@@ -6,6 +6,7 @@ import java.util.Map;
 import javax.websocket.CloseReason;
 import javax.websocket.Endpoint;
 import javax.websocket.EndpointConfig;
+import javax.websocket.MessageHandler;
 import javax.websocket.Session;
 
 import org.slf4j.Logger;
@@ -13,25 +14,30 @@ import org.slf4j.LoggerFactory;
 
 public abstract class ClientHandler extends Endpoint {
 
-    private final static Logger log = LoggerFactory.getLogger(ClientHandler.class);
+    private final static Logger LOG = LoggerFactory.getLogger(ClientHandler.class);
 
     @Override
     public void onOpen(Session session, EndpointConfig config) {
-        log.info("Websocket session {} opened.", session.getId());
-        session.addMessageHandler(String.class, message -> {
-            log.info("Message received on Websocket session {}: {}", session.getId(), message);
+        LOG.info("Websocket session {} opened.", session.getId());
+        session.addMessageHandler(new MessageHandler.Whole<String>() {
+
+            @Override
+            public void onMessage(String message) {
+                LOG.info("Message received on Websocket session {}: {}", session.getId(), message);
+            }
         });
     }
 
     @Override
     public void onClose(Session session, CloseReason reason) {
-        log.info("Websocket session {} closed.", session.getId());
+        LOG.info("Websocket session {} closed.", session.getId());
     }
 
     @Override
     public void onError(Session session, Throwable error) {
-        log.error("Error occurred on Websocket session" + session.getId(), error);
+        LOG.error("Error occurred on Websocket session" + session.getId(), error);
     }
 
-    public void beforeRequest(Map<String,List<String>> headers) {}
+    public void beforeRequest(Map<String, List<String>> headers) {
+    }
 }

client/src/main/java/timely/client/websocket/TimelyEndpointConfig.java

@@ -36,7 +36,7 @@ public class TimelyEndpointConfig implements ClientEndpointConfig {
     }
 
     @Override
-    public Map<String,Object> getUserProperties() {
+    public Map<String, Object> getUserProperties() {
         return Collections.emptyMap();
     }
 
@@ -55,7 +55,7 @@ public class TimelyEndpointConfig implements ClientEndpointConfig {
         return new Configurator() {
 
             @Override
-            public void beforeRequest(Map<String,List<String>> headers) {
+            public void beforeRequest(Map<String, List<String>> headers) {
                 super.beforeRequest(headers);
                 if (null != sessionCookie) {
                     headers.put("Cookie", Collections.singletonList(sessionCookie));

client/src/main/java/timely/client/websocket/WebSocketClient.java

@@ -10,13 +10,17 @@ import javax.net.ssl.SSLContext;
 import javax.websocket.DeploymentException;
 import javax.websocket.Session;
 
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.http.HttpResponse;
 import org.apache.http.HttpStatus;
 import org.apache.http.client.HttpResponseException;
 import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.methods.HttpRequestBase;
 import org.apache.http.client.utils.HttpClientUtils;
 import org.apache.http.cookie.Cookie;
+import org.apache.http.entity.StringEntity;
 import org.apache.http.impl.client.BasicCookieStore;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.glassfish.tyrus.client.ClientManager;
@@ -24,20 +28,21 @@ import org.glassfish.tyrus.client.ClientProperties;
 import org.glassfish.tyrus.client.SslEngineConfigurator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Preconditions;
-
+import timely.api.BasicAuthLogin;
 import timely.client.http.HttpClient;
+import timely.serialize.JsonSerializer;
 
 public class WebSocketClient implements AutoCloseable {
 
-    private static final Logger log = LoggerFactory.getLogger(WebSocketClient.class);
+    private static final Logger LOG = LoggerFactory.getLogger(WebSocketClient.class);
 
     private final String timelyHostname;
     private final int timelyHttpsPort;
     private final int timelyWssPort;
     private final boolean doLogin;
     private final boolean clientAuth;
+    private final String timelyUsername;
+    private final String timelyPassword;
     private final boolean hostVerificationEnabled;
     private final int bufferSize;
     private final SSLContext ssl;
@@ -46,27 +51,39 @@ public class WebSocketClient implements AutoCloseable {
     protected Session session = null;
     protected volatile boolean closed = true;
 
-    protected WebSocketClient(SSLContext ssl, String timelyHostname, int timelyHttpsPort, int timelyWssPort, boolean clientAuth, boolean doLogin,
-                    boolean hostVerificationEnabled, int bufferSize) {
+    protected WebSocketClient(SSLContext ssl, String timelyHostname, int timelyHttpsPort, int timelyWssPort,
+            boolean clientAuth, boolean doLogin, String timelyUsername, String timelyPassword,
+            boolean hostVerificationEnabled, int bufferSize) {
         this.ssl = ssl;
         this.timelyHostname = timelyHostname;
         this.timelyHttpsPort = timelyHttpsPort;
         this.timelyWssPort = timelyWssPort;
         this.doLogin = doLogin;
         this.clientAuth = clientAuth;
+        this.timelyUsername = timelyUsername;
+        this.timelyPassword = timelyPassword;
         this.hostVerificationEnabled = hostVerificationEnabled;
         this.bufferSize = bufferSize;
 
         Preconditions.checkNotNull(timelyHostname, "%s must be supplied", "Timely host name");
         Preconditions.checkNotNull(timelyHttpsPort, "%s must be supplied", "Timely HTTPS port");
         Preconditions.checkNotNull(timelyWssPort, "%s must be supplied", "Timely WSS port");
+
+        if (doLogin && !clientAuth) {
+            if ((StringUtils.isEmpty(timelyUsername) && !StringUtils.isEmpty(timelyPassword)
+                    || (!StringUtils.isEmpty(timelyUsername) && StringUtils.isEmpty(timelyPassword)))) {
+                throw new IllegalArgumentException("Both Timely username and password must be empty or non-empty");
+            }
+        }
     }
 
-    protected WebSocketClient(String timelyHostname, int timelyHttpsPort, int timelyWssPort, boolean clientAuth, boolean doLogin, String keyStoreFile,
-                    String keyStoreType, String keyStorePass, String trustStoreFile, String trustStoreType, String trustStorePass,
-                    boolean hostVerificationEnabled, int bufferSize) {
-        this(HttpClient.getSSLContext(trustStoreFile, trustStoreType, trustStorePass, keyStoreFile, keyStoreType, keyStorePass), timelyHostname,
-                        timelyHttpsPort, timelyWssPort, clientAuth, doLogin, hostVerificationEnabled, bufferSize);
+    protected WebSocketClient(String timelyHostname, int timelyHttpsPort, int timelyWssPort, boolean clientAuth,
+            boolean doLogin, String timelyUsername, String timelyPassword, String keyStoreFile, String keyStoreType,
+            String keyStorePass, String trustStoreFile, String trustStoreType, String trustStorePass,
+            boolean hostVerificationEnabled, int bufferSize) {
+        this(HttpClient.getSSLContext(trustStoreFile, trustStoreType, trustStorePass, keyStoreFile, keyStoreType,
+                keyStorePass), timelyHostname, timelyHttpsPort, timelyWssPort, clientAuth, doLogin, timelyUsername,
+                timelyPassword, hostVerificationEnabled, bufferSize);
     }
 
     public void open(ClientHandler clientEndpoint) throws IOException, DeploymentException, URISyntaxException {
@@ -76,16 +93,30 @@ public class WebSocketClient implements AutoCloseable {
             try (CloseableHttpClient client = HttpClient.get(ssl, cookieJar, hostVerificationEnabled, clientAuth)) {
 
                 String target = "https://" + timelyHostname + ":" + timelyHttpsPort + "/login";
+
                 HttpRequestBase request = null;
-                // HTTP GET to /login to use certificate based login
-                request = new HttpGet(target);
-                log.trace("Performing client certificate login");
+                if (StringUtils.isEmpty(timelyUsername)) {
+                    // HTTP GET to /login to use certificate based login
+                    request = new HttpGet(target);
+                    LOG.trace("Performing client certificate login");
+                } else {
+                    // HTTP POST to /login to use username/password
+                    BasicAuthLogin login = new BasicAuthLogin();
+                    login.setUsername(timelyUsername);
+                    login.setPassword(timelyPassword);
+                    String payload = JsonSerializer.getObjectMapper().writeValueAsString(login);
+                    HttpPost post = new HttpPost(target);
+                    post.setEntity(new StringEntity(payload));
+                    request = post;
+                    LOG.trace("Performing BasicAuth login");
+                }
 
                 HttpResponse response = null;
                 try {
                     response = client.execute(request);
                     if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
-                        throw new HttpResponseException(response.getStatusLine().getStatusCode(), response.getStatusLine().getReasonPhrase());
+                        throw new HttpResponseException(response.getStatusLine().getStatusCode(),
+                                response.getStatusLine().getReasonPhrase());
                     }
                     for (Cookie c : cookieJar.getCookies()) {
                         if (c.getName().equals("TSESSIONID")) {
@@ -94,7 +125,8 @@ public class WebSocketClient implements AutoCloseable {
                         }
                     }
                     if (null == sessionCookie) {
-                        throw new IllegalStateException("Unable to find TSESSIONID cookie header in Timely login response");
+                        throw new IllegalStateException(
+                                "Unable to find TSESSIONID cookie header in Timely login response");
                     }
                 } finally {
                     HttpClientUtils.closeQuietly(response);
@@ -110,14 +142,15 @@ public class WebSocketClient implements AutoCloseable {
         webSocketClient.getProperties().put(ClientProperties.SSL_ENGINE_CONFIGURATOR, sslEngine);
         webSocketClient.getProperties().put(ClientProperties.INCOMING_BUFFER_SIZE, bufferSize);
         String wssPath = "wss://" + timelyHostname + ":" + timelyWssPort + "/websocket";
-        session = webSocketClient.connectToServer(clientEndpoint, new TimelyEndpointConfig(clientEndpoint, sessionCookie), new URI(wssPath));
+        session = webSocketClient.connectToServer(clientEndpoint,
+                new TimelyEndpointConfig(clientEndpoint, sessionCookie), new URI(wssPath));
 
         final ByteBuffer pingData = ByteBuffer.allocate(0);
         webSocketClient.getScheduledExecutorService().scheduleAtFixedRate(() -> {
             try {
                 session.getBasicRemote().sendPing(pingData);
             } catch (Exception e) {
-                log.error("Error sending ping", e);
+                LOG.error("Error sending ping", e);
             }
         }, 30, 60, TimeUnit.SECONDS);
         closed = false;

client/src/main/java/timely/client/websocket/subscription/WebSocketSubscriptionClient.java

@@ -13,35 +13,37 @@ import javax.websocket.DeploymentException;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import timely.api.AddSubscription;
 import timely.api.CloseSubscription;
 import timely.api.CreateSubscription;
 import timely.api.RemoveSubscription;
-import timely.api.VersionRequest;
 import timely.client.websocket.ClientHandler;
 import timely.client.websocket.WebSocketClient;
 import timely.serialize.JsonSerializer;
 
 public class WebSocketSubscriptionClient extends WebSocketClient {
 
-    private static final Logger log = LoggerFactory.getLogger(WebSocketSubscriptionClient.class);
+    private static final Logger LOG = LoggerFactory.getLogger(WebSocketSubscriptionClient.class);
     private final String subscriptionId;
 
-    public WebSocketSubscriptionClient(SSLContext ssl, String timelyHostname, int timelyHttpsPort, int timelyWssPort, boolean clientAuth, boolean doLogin,
-                    boolean hostVerificationEnabled, int bufferSize) {
-        super(ssl, timelyHostname, timelyHttpsPort, timelyWssPort, clientAuth, doLogin, hostVerificationEnabled, bufferSize);
+    public WebSocketSubscriptionClient(SSLContext ssl, String timelyHostname, int timelyHttpsPort, int timelyWssPort,
+            boolean clientAuth, boolean doLogin, String timelyUsername, String timelyPassword,
+            boolean hostVerificationEnabled, int bufferSize) {
+        super(ssl, timelyHostname, timelyHttpsPort, timelyWssPort, clientAuth, doLogin, timelyUsername, timelyPassword,
+                hostVerificationEnabled, bufferSize);
         subscriptionId = UUID.randomUUID().toString();
-        log.trace("Created WebSocketClient with subscriptionId {}", this.subscriptionId);
+        LOG.trace("Created WebSocketClient with subscriptionId {}", this.subscriptionId);
     }
 
-    public WebSocketSubscriptionClient(String timelyHostname, int timelyHttpsPort, int timelyWssPort, boolean clientAuth, boolean doLogin, String keyStoreFile,
-                    String keyStoreType, String keyStorePass, String trustStoreFile, String trustStoreType, String trustStorePass,
-                    boolean hostVerificationEnabled, int bufferSize) {
-        super(timelyHostname, timelyHttpsPort, timelyWssPort, clientAuth, doLogin, keyStoreFile, keyStoreType, keyStorePass, trustStoreFile, trustStoreType,
-                        trustStorePass, hostVerificationEnabled, bufferSize);
+    public WebSocketSubscriptionClient(String timelyHostname, int timelyHttpsPort, int timelyWssPort,
+            boolean clientAuth, boolean doLogin, String timelyUsername, String timelyPassword, String keyStoreFile,
+            String keyStoreType, String keyStorePass, String trustStoreFile, String trustStoreType,
+            String trustStorePass, boolean hostVerificationEnabled, int bufferSize) {
+        super(timelyHostname, timelyHttpsPort, timelyWssPort, clientAuth, doLogin, timelyUsername, timelyPassword,
+                keyStoreFile, keyStoreType, keyStorePass, trustStoreFile, trustStoreType, trustStorePass,
+                hostVerificationEnabled, bufferSize);
         subscriptionId = UUID.randomUUID().toString();
-        log.trace("Created WebSocketClient with subscriptionId {}", this.subscriptionId);
+        LOG.trace("Created WebSocketClient with subscriptionId {}", this.subscriptionId);
     }
 
     @Override
@@ -61,13 +63,14 @@ public class WebSocketSubscriptionClient extends WebSocketClient {
                 session.getBasicRemote().sendText(JsonSerializer.getObjectMapper().writeValueAsString(close));
                 session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "Client called close."));
             } catch (Exception e) {
-                log.info("Unable to send close message to server: {}", e.getMessage());
+                LOG.info("Unable to send close message to server: {}", e.getMessage());
             }
         }
         super.close();
     }
 
-    public Future<Void> addSubscription(String metric, Map<String,String> tags, long startTime, long endTime, long delayTime) throws IOException {
+    public Future<Void> addSubscription(String metric, Map<String, String> tags, long startTime, long endTime,
+            long delayTime) throws IOException {
         AddSubscription add = new AddSubscription();
         add.setSubscriptionId(subscriptionId);
         add.setMetric(metric);
@@ -85,8 +88,4 @@ public class WebSocketSubscriptionClient extends WebSocketClient {
         return session.getAsyncRemote().sendText(JsonSerializer.getObjectMapper().writeValueAsString(remove));
     }
 
-    public Future<Void> requestVersion() throws Exception {
-        VersionRequest version = new VersionRequest();
-        return session.getAsyncRemote().sendText(JsonSerializer.getObjectMapper().writeValueAsString(version));
-    }
 }

client/src/main/java/timely/model/Metric.java

@@ -2,26 +2,26 @@ package timely.model;
 
 import java.io.Serializable;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
-import java.util.TreeSet;
 
 import com.fasterxml.jackson.annotation.JsonGetter;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonPropertyOrder;
 import com.fasterxml.jackson.annotation.JsonRootName;
 import com.fasterxml.jackson.annotation.JsonUnwrapped;
-
 import timely.model.parse.TagParser;
 
 /**
  * A Metric consists of a metric name, tags, and a Value
  *
- * Json has been flattened to reduce overhead. Example: {"name":"m1","timestamp":1,"measure":1.0,"tags":[{"k1":"v1"}]}
+ * Json has been flattened to reduce overhead. Example:
+ * {"name":"m1","timestamp":1,"measure":1.0,"tags":[{"k1":"v1"}]}
  *
  */
 @JsonRootName("metric")
 @JsonInclude(JsonInclude.Include.NON_NULL)
-@JsonPropertyOrder({"name"})
+@JsonPropertyOrder({ "name" })
 public class Metric implements Serializable {
 
     private static final long serialVersionUID = 1L;
@@ -79,7 +79,7 @@ public class Metric implements Serializable {
     }
 
     public void setTags(List<Tag> tags) {
-        this.tags = tags;
+        this.tags.addAll(tags);
     }
 
     public void addTag(Tag tag) {
@@ -97,7 +97,8 @@ public class Metric implements Serializable {
 
     @Override
     public String toString() {
-        return "Metric{" + "name='" + name + '\'' + ", tags=" + new TreeSet<>(tags) + ", value=" + value + '}';
+        Collections.sort(tags);
+        return "Metric{" + "name='" + name + '\'' + ", tags=" + tags + ", value=" + value + '}';
     }
 
     @Override
@@ -133,7 +134,8 @@ public class Metric implements Serializable {
     }
 
     /**
-     * Simple Builder for Metric. Note: cannot re-use a builder after build is called.
+     * Simple Builder for Metric. Note: cannot re-use a builder after build is
+     * called.
      */
     public static class Builder {
 

client/src/main/java/timely/model/ObjectSizeOf.java

@@ -2,13 +2,12 @@ package timely.model;
 
 import java.lang.reflect.Array;
 import java.lang.reflect.Field;
+import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.Stack;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import java.util.logging.Logger;
 
 /**
  * A simple interface that objects can implement to return the object size.
@@ -48,7 +47,7 @@ public interface ObjectSizeOf {
 
     public static class Sizer {
 
-        private static final Logger log = LoggerFactory.getLogger(Sizer.class);
+        private static final Logger log = Logger.getLogger(Sizer.class.getCanonicalName());
         public static final short OBJECT_OVERHEAD = 8;
         public static final short ARRAY_OVERHEAD = 12;
         public static final short REFERENCE = 4;
@@ -57,14 +56,17 @@ public interface ObjectSizeOf {
         public static final short NUMBER_SIZE = 16;
 
         /**
-         * Get the size of an object. Note that we want something relatively fast that gives us an order of magnitude here. The java Instrumentation agent
-         * mechanism is a little too costly for general use here. This will look for the ObjectSizeOf interface and if implemented on the object will use that.
+         * Get the size of an object. Note that we want something relatively fast that
+         * gives us an order of magnitude here. The java Instrumentation agent mechanism
+         * is a little too costly for general use here. This will look for the
+         * ObjectSizeOf interface and if implemented on the object will use that.
          * Otherwise it will do a simple navigation of the fields using reflection.
          *
          * @return an approximation of the object size
          */
         public static long getObjectSize(Object o, boolean useSizeInBytesMethod, boolean debug) {
-            long estSize = getObjectSize(o, new HashSet<ObjectInstance>(), new Stack<ObjectInstance>(), useSizeInBytesMethod, debug);
+            long estSize = getObjectSize(o, new HashSet<ObjectInstance>(), new Stack<ObjectInstance>(),
+                    useSizeInBytesMethod, debug);
             return estSize;
         }
 
@@ -73,7 +75,8 @@ public interface ObjectSizeOf {
             return estSize;
         }
 
-        public static long getObjectSize(Object o, Set<ObjectInstance> visited, Stack<ObjectInstance> stack, boolean useSizeInBytesMethod, boolean debug) {
+        public static long getObjectSize(Object o, Set<ObjectInstance> visited, Stack<ObjectInstance> stack,
+                boolean useSizeInBytesMethod, boolean debug) {
             long totalSize = 0;
             stack.add(new ObjectInstance(o));
             while (!stack.isEmpty()) {
@@ -84,16 +87,23 @@ public interface ObjectSizeOf {
                     visited.add(oi);
                     try {
                         if (useSizeInBytesMethod) {
-                            if (o instanceof ObjectSizeOf) {
-                                size = ((ObjectSizeOf) o).sizeInBytes();
+                            try {
+                                if (o instanceof ObjectSizeOf) {
+                                    size = ((ObjectSizeOf) o).sizeInBytes();
+                                } else {
+                                    Method sizeInBytes = o.getClass().getMethod("sizeInBytes", (Class<?>[]) null);
+                                    size = (Long) sizeInBytes.invoke(o);
+                                }
+                            } catch (Throwable t) {
+                                // ok, lets do this the hard way...
                             }
                         }
                         if (size == 0) {
                             // the hard way...
                             // do not include Class related objects or
                             // reflection objects
-                            if ((o instanceof Class) || (o instanceof ClassLoader)
-                                            || (o.getClass().getPackage() != null && o.getClass().getPackage().getName().startsWith("java.lang.reflect"))) {
+                            if ((o instanceof Class) || (o instanceof ClassLoader) || (o.getClass().getPackage() != null
+                                    && o.getClass().getPackage().getName().startsWith("java.lang.reflect"))) {
                                 size = 0;
                             } else if (o instanceof Number || o instanceof Boolean || o instanceof Character) {
                                 size = NUMBER_SIZE;
@@ -148,7 +158,7 @@ public interface ObjectSizeOf {
                             }
                         }
                     } catch (Throwable t) {
-                        log.warn("Unable to determine object size for " + o, t);
+                        log.warning("Unable to determine object size for " + o);
                     }
                     if (debug) {
                         System.out.println(o.getClass().getCanonicalName() + "=" + size);

client/src/main/java/timely/model/Tag.java

@@ -5,15 +5,13 @@ import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-import org.apache.commons.lang3.builder.CompareToBuilder;
-import org.apache.commons.lang3.builder.EqualsBuilder;
-
 import com.fasterxml.jackson.annotation.JsonAnyGetter;
 import com.fasterxml.jackson.annotation.JsonAnySetter;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonRootName;
 import com.google.common.base.Preconditions;
-
+import org.apache.commons.lang3.builder.CompareToBuilder;
+import org.apache.commons.lang3.builder.EqualsBuilder;
 import timely.model.parse.TagParser;
 
 /**
@@ -29,7 +27,8 @@ public class Tag implements Comparable<Tag>, Serializable {
     private String key;
     private String value;
 
-    public Tag() {}
+    public Tag() {
+    }
 
     public Tag(Tag other) {
         this.setKey(other.getKey());
@@ -48,7 +47,7 @@ public class Tag implements Comparable<Tag>, Serializable {
     }
 
     @JsonAnyGetter
-    public Map<String,String> get() {
+    public Map<String, String> get() {
         return Stream.of(this).collect(Collectors.toMap(Tag::getKey, Tag::getValue));
     }