Fix deadlock in tracking noncachedMetrics across processes

.gitignore

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

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.6-SNAPSHOT</version>
+    <version>1.0.1-SNAPSHOT</version>
   </parent>
   <artifactId>timely-analytics</artifactId>
   <!--
@@ -53,21 +53,21 @@
         </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>
       <plugins>
         <plugin>
-          <groupId>org.apache.maven.plugins</groupId>
           <artifactId>maven-compiler-plugin</artifactId>
           <configuration>
             <compilerId>jdt</compilerId>
@@ -76,7 +76,7 @@
             <dependency>
               <groupId>org.eclipse.tycho</groupId>
               <artifactId>tycho-compiler-jdt</artifactId>
-              <version>1.4.0</version>
+              <version>0.21.0</version>
             </dependency>
           </dependencies>
         </plugin>
@@ -168,12 +168,6 @@
                 </excludes>
               </artifactSet>
               <filters>
-                <filter>
-                  <artifact>*:*</artifact>
-                  <excludes>
-                    <exclude>**/module-info.class</exclude>
-                  </excludes>
-                </filter>
                 <filter>
                   <artifact>org.apache.flink:*</artifact>
                   <excludes>
@@ -251,14 +245,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.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();
@@ -96,27 +96,29 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
 
             @Override
             public void onOpen(Session session, EndpointConfig config) {
-                session.addMessageHandler(String.class, new MessageHandler.Whole<String>() {
+                session.addMessageHandler(new MessageHandler.Whole<String>() {
 
                     private final SimpleDateFormat formatter = new SimpleDateFormat("yyyyMMdd-HH");
                     private long lastWatermarkTime = 0L;
 
                     @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());
@@ -68,10 +67,10 @@ public class SummarizationJob {
         })
         .name("Metric Windows")
         .addSink(
-            new SocketClientSink<MetricHistogram>(jp.getTimelyHostname(), jp.getTimelyTcpPort(),
+    		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

@@ -18,7 +18,6 @@ public class SummarizationJobParameters extends GlobalJobParameters implements S
     private final int timelyTcpPort;
     private final int timelyHttpsPort;
     private final int timelyWssPort;
-    private final boolean clientAuth;
     private final boolean doLogin;
     private final String timelyUsername;
     private final String timelyPassword;
@@ -41,7 +40,6 @@ public class SummarizationJobParameters extends GlobalJobParameters implements S
         timelyTcpPort = params.getInt("timelyTcpPort", 4241);
         timelyHttpsPort = params.getInt("timelyHttpsPort", 4242);
         timelyWssPort = params.getInt("timelyWssPort", 4243);
-        clientAuth = params.getBoolean("clientAuth", false);
         doLogin = params.getBoolean("doLogin", false);
         timelyUsername = params.get("timelyUsername", null);
         timelyPassword = params.get("timelyPassword", null);
@@ -71,7 +69,6 @@ public class SummarizationJobParameters extends GlobalJobParameters implements S
         timelyTcpPort = conf.getInteger("timelyTcpPort", 4241);
         timelyHttpsPort = conf.getInteger("timelyHttpsPort", 4242);
         timelyWssPort = conf.getInteger("timelyWssPort", 4243);
-        clientAuth = conf.getBoolean("clientAuth", false);
         doLogin = conf.getBoolean("doLogin", false);
         timelyUsername = conf.getString("timelyUsername", null);
         timelyPassword = conf.getString("timelyPassword", null);
@@ -115,10 +112,6 @@ public class SummarizationJobParameters extends GlobalJobParameters implements S
         return timelyWssPort;
     }
 
-    public boolean isClientAuth() {
-        return clientAuth;
-    }
-
     public boolean isDoLogin() {
         return doLogin;
     }
@@ -184,8 +177,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/start-all.sh

@@ -1,18 +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
-
-for i in $(seq 1 $NUM_INSTANCES); do
-    echo ${BIN_DIR}/timely-balancer.sh $i
-    ${BIN_DIR}/timely-balancer.sh $i
-done

balancer/bin/stop-all.sh

@@ -1,21 +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
-
-for i in $(seq 1 $NUM_INSTANCES); do
-    PID=`jps -m | grep -E 'timely-balancer-.*-exec.jar' | grep "instance=${i}" | awk '{print $1}'`
-    if [ "$PID" != "" ]; then
-        echo "stopping timely-balancer instance ${i} [${PID}]"
-        kill -9 $PID
-    fi
-done

balancer/bin/testing/load-test-env.sh

@@ -0,0 +1,11 @@
+#!/bin/bash
+
+export TRUSTSTORE_FILE=
+export TRUSTSTORE_TYPE="jks";
+export TRUSTSTORE_PASS=
+export KEYSTORE_FILE=
+export KEYSTORE_TYPE="pkcs12";
+export KEYSTORE_PASS=
+
+export NUM_QUERY_THREADS=4
+export NUM_INGEST_THREADS=4

balancer/bin/testing/load-test-ingest.sh

@@ -0,0 +1,21 @@
+#!/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=${THIS_DIR}/../..
+TMP_DIR="${BASE_DIR}/tmp"
+CONF_DIR="${BASE_DIR}/conf"
+LIB_DIR="${BASE_DIR}/lib"
+
+. ${THIS_DIR}/load-test-env.sh
+
+export CLASSPATH="${CONF_DIR}:${THIS_DIR}:${LIB_DIR}/*"
+JVM_ARGS="-Xmx1G -Xms1G"
+
+echo "$JAVA_HOME/bin/java ${JVM_ARGS} timely.testing.MetricPutLoadTest $*"
+$JAVA_HOME/bin/java ${JVM_ARGS} timely.testing.MetricPutLoadTest $*

balancer/bin/testing/load-test-query.sh

@@ -0,0 +1,21 @@
+#!/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=${THIS_DIR}/../..
+TMP_DIR="${BASE_DIR}/tmp"
+CONF_DIR="${BASE_DIR}/conf"
+LIB_DIR="${BASE_DIR}/lib"
+
+. ${THIS_DIR}/load-test-env.sh
+
+export CLASSPATH="${CONF_DIR}:${THIS_DIR}:${LIB_DIR}/*"
+JVM_ARGS="-Xmx1G -Xms1G"
+
+echo "$JAVA_HOME/bin/java ${JVM_ARGS} timely.testing.MetricQueryLoadTest $*"
+$JAVA_HOME/bin/java ${JVM_ARGS} timely.testing.MetricQueryLoadTest $*

balancer/bin/timely-balancer-env.sh

@@ -1,27 +0,0 @@
-#!/bin/bash
-
-NUM_INSTANCES=1
-ACCUMULO_ZOOKEEPERS=localhost:2181
-HADOOP_CONF_DIR=${HADOOP_CONF_DIR}
-HADOOP_DEFAULT_FS=hdfs://localhost:8020
-TCP_BASE_PORT=4242
-HTTP_BASE_PORT=4243
-WEBSOCKET_BASE_PORT=4244
-UDP_BASE_PORT=4245
-
-HOST_IP=0.0.0.0
-HOST=$(hostname -s)
-HOST_VERIFICATION_ENABLED=true
-
-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=
-ANONYMOUS_HTTP=true
-ANONYMOUS_WEBSOCKET=true

balancer/bin/timely-balancer.sh

@@ -1,64 +1,43 @@
-#!/bin/bash
+#!/bin/bash -vx
 
-if [[ $(uname) == "Darwin" ]]; then
-    THIS_SCRIPT=$(python -c 'import os,sys; print os.path.realpath(sys.argv[1])' "$0")
+if [[ `uname` == "Darwin" ]]; then
+        THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
+        TCNATIVE_SUFFIX="jnilib"
 else
-    THIS_SCRIPT=$(readlink -f "$0")
-    HW=`uname -m`
-    OS_PATH=linux-${HW}
+        THIS_SCRIPT=`readlink -f $0`
+        TCNATIVE_SUFFIX="so"
 fi
 
-if [ -z "$1" ]; then
-    export INSTANCE="1"
-else
-    export INSTANCE="$1"
+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
+
+if [[ -e ${TMP_DIR} ]]; then
+  rm -rf ${TMP_DIR}
 fi
+mkdir ${TMP_DIR}
 
-PID=$(jps -m | grep -E 'timely-balancer-.*-exec.jar' | grep "instance=${INSTANCE}" | awk '{print $1}')
-
-if [ "$PID" == "" ]; then
-    THIS_DIR="${THIS_SCRIPT%/*}"
-    BASE_DIR=$(cd "$THIS_DIR"/.. && pwd)
-    LIB_DIR="${BASE_DIR}/lib"
-    BIN_DIR="${BASE_DIR}/bin"
-    export CONF_DIR="${BASE_DIR}/conf"
-    NATIVE_DIR="${BIN_DIR}/META-INF/native"
-
-    set -a
-    . "${BIN_DIR}/timely-balancer-env.sh"
-    set +a
-
-    # use either a value from timely-balancer-env.sh or the default
-    export LOG_DIR="${TIMELY_LOG_DIR:-${BASE_DIR}/logs}"
-
-    if [[ ! -e ${NATIVE_DIR} ]]; then
-        mkdir -p "${NATIVE_DIR}"
-    fi
-
-    if [[ ! -e ${LOG_DIR} ]]; then
-        mkdir "${LOG_DIR}"
-    fi
-
-    JVM_ARGS="-Xmx2G -Xms2G -XX:NewSize=1G -XX:MaxNewSize=1G"
-    JVM_ARGS="${JVM_ARGS} -Dlogging.config=${CONF_DIR}/log4j2.yml"
-    JVM_ARGS="${JVM_ARGS} -Dlog4j2.contextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
-    JVM_ARGS="${JVM_ARGS} -XX:+UseG1GC -XX:+UseStringDeduplication"
-    JVM_ARGS="${JVM_ARGS} -Djava.net.preferIPv4Stack=true"
-    JVM_ARGS="${JVM_ARGS} -XX:+UseNUMA"
-
-    # for linux varieties of OS, extract native libraries if not already present
-    if [[ -n "$OS_PATH" ]]; then
-        if [[ ! -d ${NATIVE_DIR} || -z "$(ls -A ${NATIVE_DIR})" ]]; then
-            pushd "${BASE_DIR}/bin" || exit
-            "${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-tcnative-boringssl-static-*.Final-"${OS_PATH}".jar META-INF/native/libnetty_tcnative_linux_"${HW}".so
-            "${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-transport-native-epoll-*-"${OS_PATH}".jar META-INF/native/libnetty_transport_native_epoll_"${HW}".so
-            popd || exit
-        fi
-        JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
-    fi
-
-    echo "${JAVA_HOME}/bin/java ${JVM_ARGS} -jar ${BIN_DIR}/timely-balancer-*-exec.jar --instance=${INSTANCE}"
-    nohup "${JAVA_HOME}"/bin/java ${JVM_ARGS} -jar "${BIN_DIR}"/timely-balancer-*-exec.jar --instance="${INSTANCE}" >> "${LOG_DIR}/timely-balancer${INSTANCE}.out" 2>&1 &
-else
-    echo "timely-balancer --instance=${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*.jar META-INF/native/libnetty_tcnative.${TCNATIVE_SUFFIX}
+$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-transport-native-epoll*.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}"
+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} -Djava.library.path=${NATIVE_DIR}/libnetty-${TCNATIVE_SUFFIX}"
+
+#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,36 +0,0 @@
-Configuration:
-  status: warn
-  monitorInterval: 60
-
-  Properties:
-    Property:
-      - name: logDir
-        value: ${env:LOG_DIR}
-      - name: PID
-        value: "????"
-      - name: LOG_PATTERN
-        value: "%d{ISO8601} %-5p [%-40.40t] [%-40.40c{1.}] %m%n%wEx"
-
-  Appenders:
-    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.xml

@@ -0,0 +1,49 @@
+<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:map key-type="java.lang.String">
+    			<beans:entry key="example.com">
+   					<util:list>
+   						<beans:value>D</beans:value>
+   						<beans:value>E</beans:value>
+   						<beans:value>F</beans:value>
+   					</util:list>
+    			</beans:entry>
+    		</util:map>
+    	</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="UserNamePasswordAuthenticationProvider" class="org.springframework.security.authentication.dao.DaoAuthenticationProvider">
+		<beans:property name="userDetailsService" ref="usernamePasswordService"/>
+	</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>
+	
+</beans:beans>

balancer/conf/timely-balancer.yml

@@ -1,149 +1,127 @@
-timely:
-  instance: ${INSTANCE}
-  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
-    cache-refresh-minutes: 5
-    authorization-url: ${AUTHORIZATION_URL}
-    allow-anonymous-http-access: ${ANONYMOUS_HTTP:true}
-    allow-anonymous-ws-access: ${ANONYMOUS_WEBSOCKET:true}
-    server-ssl:
-      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}
-    client-ssl:
-      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}
-      hostVerificationEnabled: ${HOST_VERIFICATION_ENABLED:true}
-    jwt-ssl:
-      key-store-type: ${AUTHORIZATION_CERTIFICATE_TYPE}
-      key-store-file: ${AUTHORIZATION_CERTIFICATE_FILE}
-      key-store-password: ${AUTHORIZATION_CERTIFICATE_PASSWORD}
+    allow-anonymous-access: true
+    serverSsl:
+      certificate-file:
+      key-file:
+      #key-password:
+      use-generated-keypair: true
+      trust-store-file:
+      use-openssl: true
+      #use-ciphers:
+    clientSsl:
+      keyFile:
+      keyType: pkcs12
+      #keyPassword:
+      trustStoreFile:
+      trustStoreType: jks
+      #trustStorePassword:
   cache:
     enabled: false
     metric-age-off-hours:
       default: 26
-    non-cached-metrics:
-  balancer:
-    check-server-health-interval: 10000
-    server-failures-before-down: 6
-    server-successes-before-up: 1
-    default-fs: ${HADOOP_DEFAULT_FS}
-    fs-config-resources:
-      - ${HADOOP_CONF_DIR}/core-site.xml
-      - ${HADOOP_CONF_DIR}/hdfs-site.xml
-    assignment-file: hdfs:///timely/balancer/assignments.csv
-    server:
-      ip: ${HOST_IP}
-      tcp-base-port: ${TCP_BASE_PORT}
-      num-tcp-pools: 1
-      tcp-buffer-size: 100
-      tcp-client-pool:
-        min-idle-per-key: 0
-        max-idle-per-key: 40
-        max-total-per-key: 40
-        max-total: -1
-        lifo: true
-        fairness: false
-        max-wait-millis: -1
-        min-evictable-idle-time-millis: 1800000
-        evictor-shutdown-timeout-millis: 10000
-        soft-min-evictable-idle-time-millis: -1
-        num-tests-per-eviction-run: 3
-        time-between-eviction-runs-millis: -1
-        block-when-exhausted: true
-        jmx-enabled: true
-        test-on-borrow: false
-        test-on-return: false
-      udp-base-port: ${UDP_BASE_PORT}
-      udp-client-pool:
-        min-idle-per-key: 0
-        max-idle-per-key: 40
-        max-total-per-key: 40
-        max-total: -1
-        lifo: true
-        fairness: false
-        max-wait-millis: -1
-        min-evictable-idle-time-millis: 1800000
-        evictor-shutdown-timeout-millis: 10000
-        soft-min-evictable-idle-time-millis: -1
-        num-tests-per-eviction-run: 3
-        time-between-eviction-runs-millis: -1
-        block-when-exhausted: true
-        jmx-enabled: true
-        test-on-borrow: false
-        test-on-return: false
-    http:
-      host: ${HOST}
-      ip: ${HOST_IP}
-      base-port: ${HTTP_BASE_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:
-        min-idle-per-key: 0
-        max-idle-per-key: 40
-        max-total-per-key: 40
-        max-total: -1
-        lifo: true
-        fairness: false
-        max-wait-millis: -1
-        min-evictable-idle-time-millis: 1800000
-        evictor-shutdown-timeout-millis: 10000
-        soft-min-evictable-idle-time-millis: -1
-        num-tests-per-eviction-run: 3
-        time-between-eviction-runs-millis: -1
-        block-when-exhausted: true
-        jmx-enabled: true
-        test-on-borrow: false
-        test-on-return: false
-    websocket:
-      ip: ${HOST_IP}
-      base-port: ${WEBSOCKET_BASE_PORT}
-      timeout: 60
-      subscription-lag: 120
-      scanner-batch-size: 10000
-      flush-interval-seconds: 30
-      scanner-read-ahead: 1
-      subscription-batch-size: 5000
-      incoming-buffer-size: 50000000
-      ws-client-pool:
-        min-idle-per-key: 0
-        max-idle-per-key: 40
-        max-total-per-key: 40
-        max-total: -1
-        lifo: true
-        fairness: false
-        max-wait-millis: -1
-        min-evictable-idle-time-millis: 1800000
-        evictor-shutdown-timeout-millis: 10000
-        soft-min-evictable-idle-time-millis: -1
-        num-tests-per-eviction-run: 3
-        time-between-eviction-runs-millis: -1
-        block-when-exhausted: true
-        jmx-enabled: true
-        test-on-borrow: false
-        test-on-return: false
-
-logging:
-  level:
-    ROOT: info
-    timely: debug
+    nonCachedMetrics:
+  server:
+    ip: 127.0.0.1
+    tcp-port: 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
+    udp-port: 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
+    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.6-SNAPSHOT</version>
+    <version>1.0.1-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,8 +37,12 @@
       <artifactId>commons-beanutils</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>
+      <artifactId>netty-all</artifactId>
     </dependency>
     <dependency>
       <groupId>net.sourceforge.javacsv</groupId>
@@ -51,6 +51,16 @@
     <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>
+      <artifactId>accumulo-minicluster</artifactId>
     </dependency>
     <dependency>
       <groupId>org.apache.commons</groupId>
@@ -87,19 +97,21 @@
     <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.logging.log4j</groupId>
-      <artifactId>log4j-1.2-api</artifactId>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-minicluster</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>
@@ -122,18 +134,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>
@@ -150,6 +150,10 @@
       <groupId>timely</groupId>
       <artifactId>timely-common</artifactId>
     </dependency>
+    <dependency>
+      <groupId>timely</groupId>
+      <artifactId>timely-server</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.lmax</groupId>
       <artifactId>disruptor</artifactId>
@@ -157,31 +161,9 @@
     </dependency>
     <dependency>
       <groupId>io.netty</groupId>
-      <artifactId>netty-tcnative-boringssl-static</artifactId>
-      <classifier>linux-aarch_64</classifier>
-      <scope>runtime</scope>
-    </dependency>
-    <dependency>
-      <groupId>io.netty</groupId>
-      <artifactId>netty-tcnative-boringssl-static</artifactId>
-      <classifier>linux-x86_64</classifier>
-      <scope>runtime</scope>
-    </dependency>
-    <dependency>
-      <groupId>io.netty</groupId>
-      <artifactId>netty-transport-native-epoll</artifactId>
-      <classifier>linux-aarch_64</classifier>
-      <scope>runtime</scope>
-    </dependency>
-    <dependency>
-      <groupId>io.netty</groupId>
-      <artifactId>netty-transport-native-epoll</artifactId>
-      <classifier>linux-x86_64</classifier>
-      <scope>runtime</scope>
-    </dependency>
-    <dependency>
-      <groupId>javax.activation</groupId>
-      <artifactId>activation</artifactId>
+      <artifactId>netty-tcnative</artifactId>
+      <version>${version.netty-tcnative}</version>
+      <classifier>${os.detected.classifier}</classifier>
       <scope>runtime</scope>
     </dependency>
     <dependency>
@@ -194,6 +176,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>
@@ -204,21 +191,11 @@
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>org.apache.hadoop</groupId>
-      <artifactId>hadoop-minicluster</artifactId>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>org.jsoup</groupId>
       <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>
@@ -226,16 +203,12 @@
     </dependency>
     <dependency>
       <groupId>timely</groupId>
-      <artifactId>timely-test</artifactId>
+      <artifactId>timely-common</artifactId>
       <scope>test</scope>
     </dependency>
   </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>
@@ -279,8 +252,9 @@
               <artifactItems>
                 <artifactItem>
                   <groupId>io.netty</groupId>
-                  <artifactId>netty-tcnative-boringssl-static</artifactId>
+                  <artifactId>netty-tcnative</artifactId>
                   <version>${version.netty-tcnative}</version>
+                  <classifier>${os.detected.classifier}</classifier>
                   <type>jar</type>
                   <overWrite>false</overWrite>
                   <outputDirectory>${project.build.directory}/native</outputDirectory>
@@ -295,20 +269,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>
@@ -338,15 +298,35 @@
   </build>
   <profiles>
     <profile>
-      <id>docker</id>
-      <build>
-        <plugins>
-          <plugin>
-            <groupId>com.spotify</groupId>
-            <artifactId>docker-maven-plugin</artifactId>
-          </plugin>
-        </plugins>
-      </build>
+      <id>native-epoll-linux</id>
+      <activation>
+        <os>
+          <family>!mac</family>
+        </os>
+      </activation>
+      <dependencies>
+        <dependency>
+          <groupId>io.netty</groupId>
+          <artifactId>netty-transport-native-epoll</artifactId>
+          <version>${version.netty}</version>
+          <classifier>linux-x86_64</classifier>
+        </dependency>
+      </dependencies>
+    </profile>
+    <profile>
+      <id>native-epoll</id>
+      <activation>
+        <os>
+          <family>mac</family>
+        </os>
+      </activation>
+      <dependencies>
+        <dependency>
+          <groupId>io.netty</groupId>
+          <artifactId>netty-transport-native-epoll</artifactId>
+          <version>${version.netty}</version>
+        </dependency>
+      </dependencies>
     </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,27 +16,24 @@
       </includes>
     </fileSet>
     <fileSet>
-      <directory>${project.build.directory}/lib</directory>
+      <directory>${project.build.directory}</directory>
       <outputDirectory>lib</outputDirectory>
       <includes>
-        <include>netty-tcnative-boringssl-static*linux-x86_64.jar</include>
-        <include>netty-tcnative-boringssl-static*linux-aarch_64.jar</include>
-        <include>netty-transport-native-epoll*linux-x86_64.jar</include>
-        <include>netty-transport-native-epoll*linux-aarch_64.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>
       <includes>
-        <include>**/*</include>
+        <include>*</include>
       </includes>
      </fileSet>
     <fileSet>
@@ -54,4 +51,4 @@
 	  </includes>
 	</fileSet>
   </fileSets>
-</assembly>
+</assembly>

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/ArrivalRate.java

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

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

@@ -1,34 +1,21 @@
 package timely.balancer;
 
-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 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.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.net.ssl.SSLContext;
 
-import org.apache.commons.lang3.StringUtils;
-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.SpringApplication;
-import org.springframework.context.ApplicationContext;
-
-import io.netty.bootstrap.AbstractBootstrap;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.buffer.PooledByteBufAllocator;
@@ -58,31 +45,43 @@ 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 timely.auth.JWTTokenHandler;
-import timely.balancer.configuration.BalancerHttpProperties;
-import timely.balancer.configuration.BalancerProperties;
-import timely.balancer.configuration.BalancerServerProperties;
-import timely.balancer.configuration.BalancerWebsocketProperties;
+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.builder.SpringApplicationBuilder;
+import org.springframework.context.ConfigurableApplicationContext;
+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.HealthChecker;
 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.common.component.AuthenticationService;
-import timely.common.configuration.SecurityProperties;
-import timely.common.configuration.SslClientProperties;
-import timely.common.configuration.SslServerProperties;
-import timely.common.configuration.ZookeeperProperties;
+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;
@@ -92,7 +91,7 @@ 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.WebSocketHttpCookieHandler;
 import timely.netty.websocket.WebSocketRequestDecoder;
 import timely.netty.websocket.subscription.WSTimelyExceptionHandler;
 
@@ -102,25 +101,17 @@ public class Balancer {
     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 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";
-    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 SslContext serverSslContext;
-    private SSLContext clientSSLContext;
+    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;
@@ -134,200 +125,57 @@ public class Balancer {
     protected Channel httpChannelHandle = null;
     protected Channel wsChannelHandle = null;
     protected List<Channel> udpChannelHandleList = new ArrayList<>();
-    protected int DEFAULT_EVENT_LOOP_THREADS;
+    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};
+    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")) {
-            try {
-                Runtime.Version currentVersion = Runtime.Version.parse(majMinVers);
-                Runtime.Version epollMinVersion = Runtime.Version.parse(EPOLL_MIN_VERSION);
-                return currentVersion.compareTo(epollMinVersion) > 0;
-            } catch (Exception e) {
-                log.error(e.getMessage(), e);
-            }
-        }
-        return false;
+    static {
+        DEFAULT_EVENT_LOOP_THREADS = Math.max(1,
+                SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
     }
 
-    public Balancer(ApplicationContext applicationContext, SslContext serverSslContext, SSLContext clientSSLContext,
-                    AuthenticationService authenticationService, MetricResolver metricResolver, BalancerProperties balancerProperties,
-                    ZookeeperProperties zookeeperProperties, SecurityProperties securityProperties, SslServerProperties sslServerProperties,
-                    SslClientProperties sslClientProperties, BalancerServerProperties balancerServerProperties, BalancerHttpProperties balancerHttpProperties,
-                    BalancerWebsocketProperties balancerWebsocketProperties) {
-        this.serverSslContext = serverSslContext;
-        this.clientSSLContext = clientSSLContext;
-        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));
+    public Balancer(BalancerConfiguration balancerConf) throws Exception {
+        this.balancerConfig = balancerConf;
     }
 
-    public void start() {
-        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();
-
-            if (serverSslContext instanceof OpenSslServerContext) {
-                OpenSslServerContext openssl = (OpenSslServerContext) serverSslContext;
-                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());
-            }
-            final ServerBootstrap httpServer = new ServerBootstrap();
-            httpServer.group(httpBossGroup, httpWorkerGroup);
-            httpServer.channel(channelClass);
-            httpServer.handler(new LoggingHandler());
-            HttpClientPool httpClientPool = new HttpClientPool(balancerHttpProperties, clientSSLContext, sslClientProperties);
-            httpServer.childHandler(setupHttpChannel(balancerHttpProperties, sslServerProperties, serverSslContext, 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, serverSslContext, 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 static ConfigurableApplicationContext initializeConfiguration(String[] args) {
+        return new SpringApplicationBuilder(SpringBootstrap.class).web(false).run(args);
     }
 
-    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;
+    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");
+            LOG.info("Closing tcpChannelHandle");
             channelFutures.add(tcpChannelHandle.close());
         }
 
         if (httpChannelHandle != null) {
-            log.info("Closing httpChannelHandle");
+            LOG.info("Closing httpChannelHandle");
             channelFutures.add(httpChannelHandle.close());
         }
 
         if (wsChannelHandle != null) {
-            log.info("Closing wsChannelHandle");
+            LOG.info("Closing wsChannelHandle");
             channelFutures.add(wsChannelHandle.close());
         }
 
-        int udpChannel = 1;
         for (Channel c : udpChannelHandleList) {
-            log.info("Closing udpChannelHandle #" + udpChannel++);
+            LOG.info("Closing udpChannelHandle");
             channelFutures.add(c.close());
         }
 
@@ -336,49 +184,51 @@ public class Balancer {
             try {
                 f.get();
             } catch (final Exception e) {
-                log.error("Channel:" + f.channel().config() + " -> " + e.getMessage(), e);
+                LOG.error("Error while shutting down channel: {}", f.channel().config());
+                LOG.error("{}", e.getMessage());
+                e.printStackTrace();
             }
         });
 
+        Integer quietPeriod = balancerConfig.getServer().getShutdownQuietPeriod();
         List<Future<?>> groupFutures = new ArrayList<>();
-
         if (tcpBossGroup != null) {
-            log.info("Shutting down tcpBossGroup");
+            LOG.info("Shutting down tcpBossGroup");
             groupFutures.add(tcpBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
         }
 
         if (tcpWorkerGroup != null) {
-            log.info("Shutting down tcpWorkerGroup");
+            LOG.info("Shutting down tcpWorkerGroup");
             groupFutures.add(tcpWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
         }
 
         if (httpBossGroup != null) {
-            log.info("Shutting down httpBossGroup");
+            LOG.info("Shutting down httpBossGroup");
             groupFutures.add(httpBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
         }
 
         if (httpWorkerGroup != null) {
-            log.info("Shutting down httpWorkerGroup");
+            LOG.info("Shutting down httpWorkerGroup");
             groupFutures.add(httpWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
         }
 
         if (wsBossGroup != null) {
-            log.info("Shutting down wsBossGroup");
+            LOG.info("Shutting down wsBossGroup");
             groupFutures.add(wsBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
         }
 
         if (wsWorkerGroup != null) {
-            log.info("Shutting down wsWorkerGroup");
+            LOG.info("Shutting down wsWorkerGroup");
             groupFutures.add(wsWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
         }
 
         if (udpBossGroup != null) {
-            log.info("Shutting down udpBossGroup");
+            LOG.info("Shutting down udpBossGroup");
             groupFutures.add(udpBossGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
         }
 
         if (udpWorkerGroup != null) {
-            log.info("Shutting down udpWorkerGroup");
+            LOG.info("Shutting down udpWorkerGroup");
             groupFutures.add(udpWorkerGroup.shutdownGracefully(quietPeriod, 10, TimeUnit.SECONDS));
         }
 
@@ -386,17 +236,106 @@ public class Balancer {
             try {
                 f.get();
             } catch (final Exception e) {
-                log.error("Group:" + f.toString() + " -> " + e.getMessage(), e);
+                LOG.error("Error while shutting down group: {}", f.toString());
+                LOG.error("{}", e.getMessage());
+                e.printStackTrace();
             }
         });
 
         try {
-            log.info("Closing WebSocketRequestDecoder");
+            LOG.info("Closing WebSocketRequestDecoder");
             WebSocketRequestDecoder.close();
         } catch (Exception e) {
-            log.error("Error closing WebSocketRequestDecoder during shutdown", e);
+            LOG.error("Error closing WebSocketRequestDecoder during shutdown", e);
         }
-        log.info("{} shut down.", this.getClass().getSimpleName());
+
+        if (applicationContext != null) {
+            LOG.info("Closing applicationContext");
+            applicationContext.close();
+        }
+
+        if (curatorFramework != null) {
+            curatorFramework.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) {
@@ -404,37 +343,180 @@ public class Balancer {
             try {
                 Stat stat = curatorFramework.checkExists().forPath(s);
                 if (stat == null) {
-                    curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath(s);
+                    curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT)
+                            .forPath(s);
                 }
             } catch (Exception e) {
-                log.info(e.getMessage());
+                LOG.info(e.getMessage());
             }
 
         }
     }
 
-    protected ChannelHandler setupHttpChannel(BalancerHttpProperties balancerHttpProperties, SslServerProperties sslServerProperties, SslContext sslCtx,
-                    MetricResolver metricResolver, HttpClientPool httpClientPool) {
+    public void run() throws Exception {
+
+        // start curator framework
+        RetryPolicy retryPolicy = new RetryForever(1000);
+        curatorFramework = CuratorFrameworkFactory.newClient(balancerConfig.getZooKeeper().getServers(), 30000, 1000,
+                retryPolicy);
+        curatorFramework.start();
+        ensureZkPaths(curatorFramework, zkPaths);
+
+        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));
+        }
+        HealthChecker healthChecker = new HealthChecker(this.balancerConfig, tcpClientPools.get(0));
+        this.metricResolver = new BalancedMetricResolver(curatorFramework, this.balancerConfig, healthChecker);
+
+        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, 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 three lines are necessary */
+        wsServer.option(ChannelOption.SO_SNDBUF, 1048576);
+        wsServer.option(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, 838860);
+        wsServer.option(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, 620145);
+        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.initializeConfiguration(args);
+        BalancerConfiguration balancerConf = applicationContext.getBean(BalancerConfiguration.class);
+
+        Balancer b = new Balancer(balancerConf);
+        try {
+            b.run();
+            LATCH.await();
+        } catch (final InterruptedException e) {
+            LOG.info("Server shutting down.");
+        } catch (Exception e) {
+            LOG.error("Error running server.", e);
+        } finally {
+            try {
+                b.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) {
+            protected void initChannel(SocketChannel ch) throws Exception {
 
-                ch.pipeline().addLast("ssl", new NonSslRedirectHandler(balancerHttpProperties, sslCtx));
+                ch.pipeline().addLast("ssl", new NonSslRedirectHandler(balancerConfig.getHttp(), sslCtx));
                 ch.pipeline().addLast("encoder", new HttpResponseEncoder());
-                ch.pipeline().addLast("decoder", new HttpRequestDecoder(4096, 32768, 8192, true, 128));
+                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("aggregator", new HttpObjectAggregator(8192));
                 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()));
+                ch.pipeline().addLast("queryDecoder", new timely.netty.http.HttpRequestDecoder(
+                        balancerConfig.getSecurity(), balancerConfig.getHttp()));
+                ch.pipeline().addLast("fileServer", new HttpStaticFileServerHandler());
+                ch.pipeline().addLast("login",
+                        new X509LoginRequestHandler(balancerConfig.getSecurity(), balancerConfig.getHttp()));
+                ch.pipeline().addLast("httpRelay",
+                        new HttpRelayHandler(balancerConfig, metricResolver, httpClientPool));
+                ch.pipeline().addLast("error", new TimelyExceptionHandler());
             }
         };
     }
@@ -447,7 +529,7 @@ public class Balancer {
                 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("frame", new DelimiterBasedFrameDecoder(8192, true, Delimiters.lineDelimiter()));
                 ch.pipeline().addLast("putDecoder", new UdpDecoder());
                 ch.pipeline().addLast("udpRelayHandler", new UdpRelayHandler(metricResolver, udpClientPool));
             }
@@ -460,7 +542,7 @@ public class Balancer {
             @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("frame", new DelimiterBasedFrameDecoder(8192, true, Delimiters.lineDelimiter()));
                 ch.pipeline().addLast("putDecoder", new TcpDecoder());
                 ch.pipeline().addLast("tcpRelayHandler", new TcpRelayHandler(metricResolver, tcpClientPools));
                 ch.pipeline().addLast("versionHandler", new TcpVersionHandler());
@@ -468,21 +550,23 @@ public class Balancer {
         };
     }
 
-    protected ChannelHandler setupWSChannel(BalancerWebsocketProperties balancerWebsocketProperties, SecurityProperties securityProperties, SslContext sslCtx,
-                    MetricResolver metricResolver, WsClientPool wsClientPool) {
+    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(authenticationService));
+                ch.pipeline().addLast("aggregator", new HttpObjectAggregator(8192));
+                ch.pipeline().addLast("sessionExtractor", new WebSocketHttpCookieHandler(balancerConfig.getSecurity()));
 
-                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("idle-handler",
+                        new IdleStateHandler(balancerConfig.getWebsocket().getTimeout(), 0, 0));
+                ch.pipeline().addLast("ws-protocol",
+                        new WebSocketServerProtocolHandler(WS_PATH, null, true, 65536, 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/HealthChecker.java

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

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/configuration/BalancerConfiguration.java

@@ -1,58 +1,121 @@
 package timely.balancer.configuration;
 
-import javax.net.ssl.SSLContext;
+import javax.validation.Valid;
 
-import org.apache.curator.framework.CuratorFramework;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.context.ApplicationContext;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.NestedConfigurationProperty;
+import org.springframework.stereotype.Component;
+import timely.configuration.Cache;
 
-import io.netty.handler.ssl.SslContext;
-import timely.balancer.Balancer;
-import timely.balancer.HealthChecker;
-import timely.balancer.MetricResolver;
-import timely.common.component.AuthenticationService;
-import timely.common.configuration.CacheProperties;
-import timely.common.configuration.SecurityProperties;
-import timely.common.configuration.SslClientProperties;
-import timely.common.configuration.SslServerProperties;
-import timely.common.configuration.ZookeeperProperties;
-
-@Configuration
-@EnableConfigurationProperties({BalancerProperties.class, BalancerServerProperties.class, BalancerWebsocketProperties.class, BalancerHttpProperties.class})
+@Component
+@ConfigurationProperties(prefix = "timely-balancer")
 public class BalancerConfiguration {
 
-    @Bean
-    @ConditionalOnMissingBean
-    public Balancer balancer(ApplicationContext applicationContext, @Qualifier("nettySslContext") SslContext serverSslContext,
-                    @Qualifier("outboundJDKSslContext") SSLContext clientSSLContext, AuthenticationService authenticationService, MetricResolver metricResolver,
-                    BalancerProperties balancerProperties, ZookeeperProperties zookeeperProperties, SecurityProperties securityProperties,
-                    SslServerProperties sslServerProperties, SslClientProperties sslClientProperties, BalancerServerProperties balancerServerProperties,
-                    BalancerHttpProperties balancerHttpProperties, BalancerWebsocketProperties balancerWebsocketProperties) {
-        Balancer balancer = new Balancer(applicationContext, serverSslContext, clientSSLContext, authenticationService, metricResolver, balancerProperties,
-                        zookeeperProperties, securityProperties, sslServerProperties, sslClientProperties, balancerServerProperties, balancerHttpProperties,
-                        balancerWebsocketProperties);
-        balancer.start();
-        return balancer;
+    @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;
+
+    public ZooKeeper getZooKeeper() {
+        return zooKeeper;
     }
 
-    @Bean
-    @ConditionalOnMissingBean
-    public HealthChecker healthChecker(BalancerProperties balancerProperties, BalancerServerProperties balancerServerProperties) {
-        HealthChecker healthChecker = new HealthChecker(balancerProperties, balancerServerProperties);
-        healthChecker.start();
-        return healthChecker;
+    public void setLoginRequired(boolean loginRequired) {
+        this.loginRequired = loginRequired;
     }
 
-    @Bean
-    @ConditionalOnMissingBean
-    public MetricResolver metricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
-                    HealthChecker healthChecker) throws Exception {
-        MetricResolver metricResolver = new MetricResolver(curatorFramework, balancerProperties, cacheProperties, healthChecker);
-        metricResolver.start();
-        return metricResolver;
+    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;
     }
 }

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

@@ -0,0 +1,18 @@
+package timely.balancer.configuration;
+
+import javax.validation.Valid;
+
+import org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig;
+import org.springframework.boot.context.properties.NestedConfigurationProperty;
+import timely.configuration.Http;
+
+public class BalancerHttp extends Http {
+
+    @Valid
+    @NestedConfigurationProperty
+    private GenericKeyedObjectPoolConfig httpClientPool = new GenericKeyedObjectPoolConfig();
+
+    public GenericKeyedObjectPoolConfig getHttpClientPool() {
+        return httpClientPool;
+    }
+}

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

@@ -1,24 +0,0 @@
-package timely.balancer.configuration;
-
-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.common.configuration.HttpProperties;
-
-@RefreshScope
-@ConfigurationProperties(prefix = "timely.balancer.http")
-public class BalancerHttpProperties extends HttpProperties {
-
-    @Valid
-    @NestedConfigurationProperty
-    private GenericKeyedObjectPoolConfig<CloseableHttpClient> httpClientPool = new GenericKeyedObjectPoolConfig<>();
-
-    public GenericKeyedObjectPoolConfig<CloseableHttpClient> getHttpClientPool() {
-        return httpClientPool;
-    }
-}

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

@@ -1,93 +0,0 @@
-package timely.balancer.configuration;
-
-import java.util.List;
-
-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 String defaultFs;
-
-    private List<String> fsConfigResources;
-
-    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 String getDefaultFs() {
-        return defaultFs;
-    }
-
-    public void setDefaultFs(String defaultFs) {
-        this.defaultFs = defaultFs;
-    }
-
-    public List<String> getFsConfigResources() {
-        return fsConfigResources;
-    }
-
-    public void setFsConfigResources(List<String> fsConfigResources) {
-        this.fsConfigResources = fsConfigResources;
-    }
-
-    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,35 +3,28 @@ 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.configuration.Server;
 
-import timely.client.tcp.TcpClient;
-import timely.client.udp.UdpClient;
-import timely.common.configuration.ServerProperties;
-
-@RefreshScope
-@ConfigurationProperties(prefix = "timely.balancer.server")
-public class BalancerServerProperties extends ServerProperties {
+public class BalancerServer extends Server {
 
     @Valid
     @NestedConfigurationProperty
-    private GenericKeyedObjectPoolConfig<UdpClient> udpClientPool = new GenericKeyedObjectPoolConfig<>();
+    private GenericKeyedObjectPoolConfig udpClientPool = new GenericKeyedObjectPoolConfig();
 
     @Valid
     @NestedConfigurationProperty
-    private GenericKeyedObjectPoolConfig<TcpClient> tcpClientPool = new GenericKeyedObjectPoolConfig<>();
+    private GenericKeyedObjectPoolConfig tcpClientPool = new GenericKeyedObjectPoolConfig();
 
     private int numTcpPools = 1;
 
     private int tcpBufferSize = -1;
 
-    public GenericKeyedObjectPoolConfig<UdpClient> getUdpClientPool() {
+    public GenericKeyedObjectPoolConfig getUdpClientPool() {
         return udpClientPool;
     }
 
-    public GenericKeyedObjectPoolConfig<TcpClient> getTcpClientPool() {
+    public GenericKeyedObjectPoolConfig getTcpClientPool() {
         return tcpClientPool;
     }
 

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

@@ -0,0 +1,18 @@
+package timely.balancer.configuration;
+
+import javax.validation.Valid;
+
+import org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig;
+import org.springframework.boot.context.properties.NestedConfigurationProperty;
+import timely.configuration.Websocket;
+
+public class BalancerWebsocket extends Websocket {
+
+    @Valid
+    @NestedConfigurationProperty
+    private GenericKeyedObjectPoolConfig wsClientPool = new GenericKeyedObjectPoolConfig();
+
+    public GenericKeyedObjectPoolConfig getWsClientPool() {
+        return wsClientPool;
+    }
+}

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

@@ -1,34 +0,0 @@
-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;
-
-@RefreshScope
-@ConfigurationProperties(prefix = "timely.balancer.websocket")
-public class BalancerWebsocketProperties extends WebsocketProperties {
-
-    @Valid
-    @NestedConfigurationProperty
-    private GenericKeyedObjectPoolConfig<WebSocketSubscriptionClient> wsClientPool = new GenericKeyedObjectPoolConfig<>();
-
-    public GenericKeyedObjectPoolConfig<WebSocketSubscriptionClient> getWsClientPool() {
-        return wsClientPool;
-    }
-
-    private int incomingBufferSize = 50000000;
-
-    public int getIncomingBufferSize() {
-        return incomingBufferSize;
-    }
-
-    public void setIncomingBufferSize(int incomingBufferSize) {
-        this.incomingBufferSize = incomingBufferSize;
-    }
-}

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

@@ -0,0 +1,60 @@
+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;
+
+    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;
+
+    }
+}

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

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

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

@@ -0,0 +1,14 @@
+package timely.balancer.configuration;
+
+public class ZooKeeper {
+
+    private String servers;
+
+    public String getServers() {
+        return servers;
+    }
+
+    public void setServers(String servers) {
+        this.servers = servers;
+    }
+}

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

@@ -1,17 +1,19 @@
 package timely.balancer.connection;
 
+import java.util.Timer;
+
 import org.apache.commons.lang3.builder.EqualsBuilder;
 import org.apache.commons.lang3.builder.HashCodeBuilder;
 import org.apache.commons.lang3.builder.ToStringBuilder;
 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);
+    static private Timer arrivalRateTimer = new Timer("ArrivalRateTimerTBH", true);
 
     private String host;
     private int tcpPort;
@@ -19,30 +21,41 @@ 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;
+    private ArrivalRate arrivalRate = new ArrivalRate(arrivalRateTimer);
 
-    private TimelyBalancedHost() {
+    public TimelyBalancedHost() {
 
     }
 
-    private TimelyBalancedHost(String host, int tcpPort, int httpPort, int wsPort, int udpPort, ArrivalRate arrivalRate) {
+    public TimelyBalancedHost(String host, int tcpPort, int httpPort, int wsPort, int udpPort) {
         this.host = host;
         this.tcpPort = tcpPort;
         this.httpPort = httpPort;
         this.wsPort = wsPort;
         this.udpPort = udpPort;
-        this.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 TimelyBalancedHost(String host, int basePort) {
+        this.host = host;
+        this.tcpPort = basePort;
+        this.httpPort = basePort + 1;
+        this.wsPort = basePort + 2;
+        this.udpPort = basePort + 3;
     }
 
-    public void setBalancerProperties(BalancerProperties balancerProperties) {
-        this.balancerProperties = balancerProperties;
+    static public TimelyBalancedHost of(String host, int tcpPort, int httpPort, int wsPort, int udpPort) {
+        return new TimelyBalancedHost(host, tcpPort, httpPort, wsPort, udpPort);
+    }
+
+    static public TimelyBalancedHost of(String host, int basePort) {
+        return new TimelyBalancedHost(host, basePort);
+    }
+
+    public void setBalancerConfig(BalancerConfiguration balancerConfig) {
+        this.balancerConfig = balancerConfig;
     }
 
     synchronized public boolean isUp() {
@@ -50,38 +63,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);
                 }
             }
         }
@@ -128,17 +141,11 @@ public class TimelyBalancedHost {
     }
 
     public void arrived() {
-        if (this.arrivalRate != null) {
-            this.arrivalRate.arrived();
-        }
+        arrivalRate.arrived();
     }
 
     public double getArrivalRate() {
-        if (this.arrivalRate == null) {
-            return 0;
-        } else {
-            return this.arrivalRate.getRate();
-        }
+        return arrivalRate.getRate();
     }
 
     @Override

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

@@ -5,27 +5,22 @@ 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 org.apache.http.impl.client.BasicCookieStore;
 import org.apache.http.impl.client.CloseableHttpClient;
-
 import timely.balancer.connection.TimelyBalancedHost;
-import timely.client.http.HttpClient;
-import timely.common.configuration.SslClientProperties;
 
-public class HttpClientFactory implements KeyedPooledObjectFactory<TimelyBalancedHost,CloseableHttpClient> {
+public class HttpClientFactory implements KeyedPooledObjectFactory<TimelyBalancedHost, CloseableHttpClient> {
 
     private final SSLContext sslContext;
-    private SslClientProperties sslClientProperties;
 
-    public HttpClientFactory(SSLContext sslContext, SslClientProperties sslClientProperties) {
+    public HttpClientFactory(SSLContext sslContext) {
         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<>(
-                        HttpClient.get(this.sslContext, null, sslClientProperties.isHostVerificationEnabled(), sslClientProperties.isUseClientCert()));
+        BasicCookieStore cookieJar = new BasicCookieStore();
+        return new DefaultPooledObject<>(timely.client.http.HttpClient.get(this.sslContext, cookieJar, false));
     }
 
     @Override

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

@@ -4,14 +4,16 @@ 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 org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import timely.balancer.configuration.BalancerConfiguration;
 import timely.balancer.connection.TimelyBalancedHost;
-import timely.common.configuration.SslClientProperties;
 
-public class HttpClientPool extends GenericKeyedObjectPool<TimelyBalancedHost,CloseableHttpClient> {
+public class HttpClientPool extends GenericKeyedObjectPool<TimelyBalancedHost, CloseableHttpClient> {
 
-    public HttpClientPool(BalancerHttpProperties http, SSLContext sslContext, SslClientProperties sslClientProperties) {
-        super(new HttpClientFactory(sslContext, sslClientProperties), http.getHttpClientPool());
+    private static final Logger LOG = LoggerFactory.getLogger(HttpClientPool.class);
+
+    public HttpClientPool(BalancerConfiguration config, SSLContext sslContext) {
+        super(new HttpClientFactory(sslContext), config.getHttp().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 final boolean doLogin;
+    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.sslContext = sslContext;
+        this.doLogin = balancerConfig.isLoginRequired();
+        this.balancerConfig = balancerConfig;
     }
 
     @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().getSubscriptionBatchSize() * 500;
+        WebSocketSubscriptionClient client = new WebSocketSubscriptionClient(sslContext, k.getHost(), k.getHttpPort(),
+                k.getWsPort(), doLogin, "", "", false, 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,53 @@
+package timely.balancer.healthcheck;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Timer;
+import java.util.TimerTask;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import timely.balancer.configuration.BalancerConfiguration;
+import timely.balancer.connection.TimelyBalancedHost;
+import timely.balancer.connection.tcp.TcpClientPool;
+
+public class HealthChecker {
+
+    private static final Logger LOG = LoggerFactory.getLogger(HealthChecker.class);
+    private List<TimelyBalancedHost> timelyHosts = new ArrayList<>();
+    private Timer timer;
+    final private TCPServerHealthCheck tcpServerHealthCheck;
+    private TimerTask check = new TimerTask() {
+
+        @Override
+        public void run() {
+            synchronized (timelyHosts) {
+                for (TimelyBalancedHost h : timelyHosts) {
+                    try {
+                        if (tcpServerHealthCheck.isServerHealthy(h)) {
+                            h.reportSuccess();
+                        } else {
+                            h.reportFailure();
+                        }
+                    } catch (Exception e) {
+                        LOG.error(e.getMessage(), e);
+                        h.reportFailure();
+                    }
+                }
+            }
+        }
+    };
+
+    public HealthChecker(BalancerConfiguration config, TcpClientPool tcpClientPool) {
+        this.tcpServerHealthCheck = new TCPServerHealthCheck(tcpClientPool);
+        this.timer = new Timer("HealthCheckerThread", true);
+        this.timer.scheduleAtFixedRate(check, 0, config.getCheckServerHealthInterval());
+    }
+
+    public void setTimelyHosts(List<TimelyBalancedHost> timelyHosts) {
+        synchronized (timelyHosts) {
+            this.timelyHosts.clear();
+            this.timelyHosts.addAll(timelyHosts);
+        }
+    }
+}

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

@@ -1,8 +1,9 @@
 package timely.balancer.netty.http;
 
-import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH;
+import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_LENGTH;
 
 import java.io.ByteArrayOutputStream;
+import java.net.URI;
 import java.net.URLEncoder;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
@@ -11,20 +12,6 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 
-import org.apache.http.Header;
-import org.apache.http.client.methods.CloseableHttpResponse;
-import org.apache.http.client.methods.HttpGet;
-import org.apache.http.client.methods.HttpPost;
-import org.apache.http.client.methods.HttpUriRequest;
-import org.apache.http.entity.StringEntity;
-import org.apache.http.impl.client.CloseableHttpClient;
-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;
@@ -35,26 +22,37 @@ 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 org.apache.http.Header;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.message.BasicHeader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.util.StreamUtils;
 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.MetricResolver;
+import timely.balancer.configuration.BalancerConfiguration;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.balancer.connection.http.HttpClientPool;
+import timely.balancer.resolver.MetricResolver;
 import timely.netty.http.TimelyHttpHandler;
 
 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 BalancerConfiguration balancerConf;
     private final HttpClientPool httpClientPool;
     private MetricResolver metricResolver;
 
-    public HttpRelayHandler(MetricResolver metricResolver, HttpClientPool httpClientPool) {
+    public HttpRelayHandler(BalancerConfiguration balancerConf, MetricResolver metricResolver,
+            HttpClientPool httpClientPool) {
+        this.balancerConf = balancerConf;
         this.metricResolver = metricResolver;
         this.httpClientPool = httpClientPool;
     }
@@ -69,7 +67,7 @@ public class HttpRelayHandler extends SimpleChannelInboundHandler<HttpRequest> i
         try {
             try {
                 request = msg.getHttpRequest();
-                String originalURI = request.uri();
+                String originalURI = request.getUri();
                 originalURI = encodeURI(originalURI);
 
                 String metric = null;
@@ -88,22 +86,14 @@ public class HttpRelayHandler extends SimpleChannelInboundHandler<HttpRequest> i
                 } else {
                     k = metricResolver.getHostPortKey(null);
                 }
+                k = metricResolver.getHostPortKey(metric);
                 client = httpClientPool.borrowObject(k);
 
-                Multimap<String,String> headers = HttpHeaderUtils.toMultimap(request.headers());
-                if (msg instanceof AuthenticatedRequest) {
-                    AuthenticatedRequest authenticatedRequest = (AuthenticatedRequest) msg;
-                    TimelyAuthenticationToken token = authenticatedRequest.getToken();
-                    if (token != null) {
-                        if (token.getClientCert() != null) {
-                            ProxiedEntityUtils.addProxyHeaders(headers, token.getClientCert());
-                        }
-                    }
-                }
-
+                List<Map.Entry<String, String>> headerList = request.headers().entries();
                 List<Header> relayedHeaderList = new ArrayList<>();
-                for (Map.Entry<String,String> h : headers.entries()) {
-                    if (!h.getKey().equals(CONTENT_LENGTH.toString())) {
+
+                for (Map.Entry<String, String> h : headerList) {
+                    if (!h.getKey().equals(CONTENT_LENGTH)) {
                         relayedHeaderList.add(new BasicHeader(h.getKey(), h.getValue()));
                     }
                 }
@@ -111,14 +101,14 @@ public class HttpRelayHandler extends SimpleChannelInboundHandler<HttpRequest> i
                 relayedHeaderList.toArray(headerArray);
 
                 String relayURI = "https://" + k.getHost() + ":" + k.getHttpPort() + originalURI;
-                if (request.method().equals(HttpMethod.GET)) {
+                if (request.getMethod().equals(HttpMethod.GET)) {
                     HttpUriRequest relayedRequest = new HttpGet(relayURI);
                     if (headerArray.length > 0) {
                         relayedRequest.setHeaders(headerArray);
                     }
                     relayedResponse = client.execute(relayedRequest);
 
-                } else if (request.method().equals(HttpMethod.POST)) {
+                } else if (request.getMethod().equals(HttpMethod.POST)) {
                     HttpPost relayedRequest = new HttpPost(relayURI);
 
                     String content = request.content().toString(StandardCharsets.UTF_8);
@@ -132,33 +122,39 @@ 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 {
+                // buf = JsonUtil.getObjectMapper().writeValueAsBytes(msg);
                 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());
                 }
+                // response.headers().set(HttpHeaders.Names.CONTENT_TYPE,
+                // Constants.JSON_TYPE);
                 response.headers().set(CONTENT_LENGTH, response.content().readableBytes());
                 sendResponse(ctx, response);
             }
@@ -169,14 +165,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().getUri());
             }
         }
     }
 
     static public class MyByteArrayOutputStream extends ByteArrayOutputStream {
 
-        public MyByteArrayOutputStream() {}
+        public MyByteArrayOutputStream() {
+        }
 
         public MyByteArrayOutputStream(int size) {
             super(size);
@@ -200,4 +197,13 @@ public class HttpRelayHandler extends SimpleChannelInboundHandler<HttpRequest> i
         }
         return s;
     }
+
+    public static void main(String[] args) {
+
+        String s = "https://127.0.0.1:4243/api/query?start=1356998400000&end=1528398304000&m=sum:rate{false,100,0}:sys.thermal.gauge{class=aws}";
+        URI uri = URI.create(encodeURI(s));
+
+        System.out.println(uri.toString());
+    }
+
 }

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.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.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

@@ -1,70 +1,41 @@
 package timely.balancer.netty.ws;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 import javax.websocket.CloseReason;
 import javax.websocket.EndpointConfig;
+import javax.websocket.MessageHandler;
 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;
 
 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;
 
-    public WsClientHandler(ChannelHandlerContext ctx, TimelyAuthenticationToken token, int pingRate) {
+    public WsClientHandler(ChannelHandlerContext ctx, int pingRate) {
         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) {
-        if (token.getClientCert() != null) {
-            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();
-        for (String s : originalRequestHeaders.keySet()) {
-            if (!headers.containsKey(s)) {
-                // add pre-existing values if key does not exist in proxyRequest
-                headers.put(s, new ArrayList<>(originalRequestHeaders.get(s)));
-            }
-        }
-    }
-
     @Override
     public void onOpen(Session session, EndpointConfig config) {
-        session.addMessageHandler(String.class, message -> {
+        session.addMessageHandler((MessageHandler.Whole<String>) 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 +43,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 +56,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,10 @@ 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 +16,31 @@ 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.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 +50,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;
     }
@@ -67,10 +64,6 @@ public class WsRelayHandler extends SimpleChannelInboundHandler<SubscriptionRequ
         String subscriptionId;
         try {
             String metric = null;
-
-            Multimap<String,String> headers = msg.getRequestHeaders();
-            ProxiedEntityUtils.addProxyHeaders(headers, msg.getToken().getClientCert());
-
             if (msg instanceof CreateSubscription) {
                 CreateSubscription create = (CreateSubscription) msg;
                 final String currentSubscriptionId = create.getSubscriptionId();
@@ -84,13 +77,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 +102,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 +118,13 @@ 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, (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 +148,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 +176,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 +194,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/MetricResolver.java

@@ -0,0 +1,10 @@
+package timely.balancer.resolver;
+
+import timely.balancer.connection.TimelyBalancedHost;
+
+public interface MetricResolver {
+
+    public TimelyBalancedHost getHostPortKeyIngest(String metric);
+
+    public TimelyBalancedHost getHostPortKey(String metric);
+}

balancer/src/main/resources/banner.txt

@@ -1,11 +0,0 @@
-===============================================================================
-  _______ _                _         ____        _
- |__   __(_)              | |       |  _ \      | |
-    | |   _ _ __ ___   ___| |_   _  | |_) | __ _| | __ _ _ __   ___ ___ _ __
-    | |  | | '_ ` _ \ / _ \ | | | | |  _ < / _` | |/ _` | '_ \ / __/ _ \ '__|
-    | |  | | | | | | |  __/ | |_| | | |_) | (_| | | (_| | | | | (_|  __/ |
-    |_|  |_|_| |_| |_|\___|_|\__, | |____/ \__,_|_|\__,_|_| |_|\___\___|_|
-                              __/ |
-                             |___/
-===============================================================================
-TimelyBalancer ${application.formatted-version}

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

@@ -10,12 +10,4 @@
     <Method name="getBuf" />
     <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>

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

@@ -1,56 +0,0 @@
-package timely.balancer.configuration;
-
-import javax.net.ssl.SSLContext;
-
-import org.apache.curator.framework.CuratorFramework;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.context.ApplicationContext;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.ComponentScan;
-import org.springframework.context.annotation.Configuration;
-
-import io.netty.handler.ssl.SslContext;
-import timely.balancer.HealthChecker;
-import timely.balancer.MetricResolver;
-import timely.balancer.test.TestBalancer;
-import timely.balancer.test.TestHealthChecker;
-import timely.balancer.test.TestMetricResolver;
-import timely.common.component.AuthenticationService;
-import timely.common.configuration.CacheProperties;
-import timely.common.configuration.SecurityProperties;
-import timely.common.configuration.SslClientProperties;
-import timely.common.configuration.SslServerProperties;
-import timely.common.configuration.ZookeeperProperties;
-
-@Configuration
-@ComponentScan(basePackages = {"timely.balancer", "timely.common", "timely.test"})
-public class TestBalancerConfiguration {
-
-    @Bean
-    public TestBalancer balancer(ApplicationContext applicationContext, @Qualifier("nettySslContext") SslContext serverSslContext,
-                    @Qualifier("outboundJDKSslContext") SSLContext clientSSLContext, AuthenticationService authenticationService, MetricResolver metricResolver,
-                    BalancerProperties balancerProperties, ZookeeperProperties zookeeperProperties, SecurityProperties securityProperties,
-                    SslServerProperties sslServerProperties, SslClientProperties sslClientProperties, BalancerServerProperties balancerServerProperties,
-                    BalancerHttpProperties balancerHttpProperties, BalancerWebsocketProperties balancerWebsocketProperties) {
-        TestBalancer balancer = new TestBalancer(applicationContext, serverSslContext, clientSSLContext, authenticationService, metricResolver,
-                        balancerProperties, zookeeperProperties, securityProperties, sslServerProperties, sslClientProperties, balancerServerProperties,
-                        balancerHttpProperties, balancerWebsocketProperties);
-        balancer.start();
-        return balancer;
-    }
-
-    @Bean
-    public TestHealthChecker healthChecker(BalancerProperties balancerProperties, BalancerServerProperties balancerServerProperties) {
-        TestHealthChecker healthChecker = new TestHealthChecker(balancerProperties, balancerServerProperties);
-        healthChecker.start();
-        return healthChecker;
-    }
-
-    @Bean
-    public TestMetricResolver metricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
-                    HealthChecker healthChecker) throws Exception {
-        TestMetricResolver metricResolver = new TestMetricResolver(curatorFramework, balancerProperties, cacheProperties, healthChecker);
-        metricResolver.start();
-        return metricResolver;
-    }
-}

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/integration/MetricResolverIT.java

@@ -1,138 +0,0 @@
-package timely.balancer.integration;
-
-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.connection.TimelyBalancedHost;
-import timely.balancer.test.TestHealthChecker;
-import timely.balancer.test.TestMetricResolver;
-import timely.test.IntegrationTest;
-import timely.test.TimelyTestRule;
-
-@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 TimelyTestRule testRule;
-
-    @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/TestBalancer.java

@@ -1,36 +0,0 @@
-package timely.balancer.test;
-
-import javax.net.ssl.SSLContext;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.context.ApplicationContext;
-
-import io.netty.handler.ssl.SslContext;
-import timely.balancer.Balancer;
-import timely.balancer.MetricResolver;
-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;
-
-public class TestBalancer extends Balancer {
-    private static final Logger log = LoggerFactory.getLogger(TestBalancer.class);
-
-    public TestBalancer(ApplicationContext applicationContext, SslContext serverSslContext, SSLContext clientSSLContext,
-                    AuthenticationService authenticationService, MetricResolver metricResolver, BalancerProperties balancerProperties,
-                    ZookeeperProperties zookeeperProperties, SecurityProperties securityProperties, SslServerProperties sslServerProperties,
-                    SslClientProperties sslClientProperties, BalancerServerProperties balancerServerProperties, BalancerHttpProperties balancerHttpProperties,
-                    BalancerWebsocketProperties balancerWebsocketProperties) {
-        super(applicationContext, serverSslContext, clientSSLContext, 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/test/TestHealthChecker.java

@@ -1,19 +0,0 @@
-package timely.balancer.test;
-
-import java.util.List;
-
-import timely.balancer.HealthChecker;
-import timely.balancer.configuration.BalancerProperties;
-import timely.balancer.configuration.BalancerServerProperties;
-import timely.balancer.connection.TimelyBalancedHost;
-
-public class TestHealthChecker extends HealthChecker {
-
-    public TestHealthChecker(BalancerProperties balancerProperties, BalancerServerProperties balancerServerProperties) {
-        super(balancerProperties, balancerServerProperties);
-    }
-
-    public List<TimelyBalancedHost> getTimelyHosts() {
-        return this.timelyHosts;
-    }
-}

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

@@ -1,44 +0,0 @@
-package timely.balancer.test;
-
-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 timely.balancer.HealthChecker;
-import timely.balancer.MetricResolver;
-import timely.balancer.configuration.BalancerProperties;
-import timely.balancer.connection.TimelyBalancedHost;
-import timely.common.configuration.CacheProperties;
-
-public class TestMetricResolver extends MetricResolver {
-
-    private static final Logger log = LoggerFactory.getLogger(TestMetricResolver.class);
-
-    public TestMetricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
-                    HealthChecker healthChecker) throws Exception {
-        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/test/TestTimelyAuthenticationManager.java

@@ -1,20 +0,0 @@
-package timely.balancer.test;
-
-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/resources/config/application.yml

@@ -1,159 +0,0 @@
-logging:
-  level:
-    ROOT: info
-    timely: debug
-    org.apache.zookeeper: error
-
-timely:
-  test: true
-  instance: 1
-  accumulo:
-    instance-name: default
-    username: root
-    password: secret
-  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
-    authorized-users:
-      - subject-dn: cn=example.com
-        roles: G,H,I
-        auths: D,E,F
-        user-type: USER
-      - subject-dn: cn=test a. user, ou=example developers, o=example corp, c=us
-        issuer-dn: cn=example corp ca, o=example corp, c=us
-        roles: G,H,I
-        auths: A,B,C
-        user-type: USER
-    server-ssl:
-      key-store-type: PKCS12
-      key-store-file: classpath:ssl/testServer.p12
-      key-store-password: ChangeIt
-      trust-store-type: JKS
-      trust-store-file: classpath:ssl/ca.jks
-      trust-store-password: ChangeIt
-      use-generated-keypair: false
-      ignore-ssl-handshake-errors: true
-      use-openssl: true
-    client-ssl:
-      key-store-type: PKCS12
-      key-store-file: classpath:ssl/testUser.p12
-      key-store-password: ChangeIt
-      trust-store-type: JKS
-      trust-store-file: classpath:ssl/ca.jks
-      trust-store-password: ChangeIt
-  cache:
-    enabled: false
-    metric-age-off-hours:
-      default: 26
-    non-cached-metrics:
-  balancer:
-    check-server-health-interval: 10000
-    server-failures-before-down: 6
-    server-successes-before-up: 1
-    assignment-file: file://${java.io.tmpdir}/timely/balancer/assignments.csv
-    server:
-      ip: 127.0.0.1
-      tcp-base-port: 4242
-      num-tcp-pools: 1
-      tcp-buffer-size: 100
-      tcp-client-pool:
-        min-idle-per-key: 0
-        max-idle-per-key: 40
-        max-total-per-key: 40
-        max-total: -1
-        lifo: true
-        fairness: false
-        max-wait-millis: -1
-        min-evictable-idle-time-millis: 1800000
-        evictor-shutdown-timeout-millis: 10000
-        soft-min-evictable-idle-time-millis: -1
-        num-tests-per-eviction-run: 3
-        time-between-eviction-runs-millis: -1
-        block-when-exhausted: true
-        jmx-enabled: true
-        test-on-borrow: false
-        test-on-return: false
-      udp-base-port: 4245
-      udp-client-pool:
-        min-idle-per-key: 0
-        max-idle-per-key: 40
-        max-total-per-key: 40
-        max-total: -1
-        lifo: true
-        fairness: false
-        max-wait-millis: -1
-        min-evictable-idle-time-millis: 1800000
-        evictor-shutdown-timeout-millis: 10000
-        soft-min-evictable-idle-time-millis: -1
-        num-tests-per-eviction-run: 3
-        time-between-eviction-runs-millis: -1
-        block-when-exhausted: true
-        jmx-enabled: true
-        test-on-borrow: false
-        test-on-return: false
-    http:
-      host: localhost
-      ip: 127.0.0.1
-      base-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
-      http-client-pool:
-        min-idle-per-key: 0
-        max-idle-per-key: 40
-        max-total-per-key: 40
-        max-total: -1
-        lifo: true
-        fairness: false
-        max-wait-millis: -1
-        min-evictable-idle-time-millis: 1800000
-        evictor-shutdown-timeout-millis: 10000
-        soft-min-evictable-idle-time-millis: -1
-        num-tests-per-eviction-run: 3
-        time-between-eviction-runs-millis: -1
-        block-when-exhausted: true
-        jmx-enabled: true
-        test-on-borrow: false
-        test-on-return: false
-    websocket:
-      ip: 127.0.0.1
-      base-port: 5244
-      timeout: 60
-      subscription-lag: 120
-      scanner-batch-size: 10000
-      flush-interval-seconds: 30
-      scanner-read-ahead: 1
-      subscription-batch-size: 5000
-      incoming-buffer-size: 50000000
-      ws-client-pool:
-        min-idle-per-key: 0
-        max-idle-per-key: 40
-        max-total-per-key: 40
-        max-total: -1
-        lifo: true
-        fairness: false
-        max-wait-millis: -1
-        min-evictable-idle-time-millis: 1800000
-        evictor-shutdown-timeout-millis: 10000
-        soft-min-evictable-idle-time-millis: -1
-        num-tests-per-eviction-run: 3
-        time-between-eviction-runs-millis: -1
-        block-when-exhausted: true
-        jmx-enabled: true
-        test-on-borrow: false
-        test-on-return: false

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

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

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

balancer/src/test/resources/ssl/ca.cert

@@ -1,23 +0,0 @@
-Bag Attributes: <No Attributes>
-subject=/C=US/O=Example Corp/CN=EXAMPLE CORP CA
-issuer=/C=US/O=Example Corp/CN=EXAMPLE CORP CA
------BEGIN CERTIFICATE-----
-MIIDTjCCAjagAwIBAgIURUj8l+ms/i9jeabIBBi19KxEhAgwDQYJKoZIhvcNAQEL
-BQAwPjELMAkGA1UEBhMCVVMxFTATBgNVBAoTDEV4YW1wbGUgQ29ycDEYMBYGA1UE
-AxMPRVhBTVBMRSBDT1JQIENBMCAXDTIzMDIwMTIzMTUwMFoYDzIxMjMwMTA4MjMx
-NTAwWjA+MQswCQYDVQQGEwJVUzEVMBMGA1UEChMMRXhhbXBsZSBDb3JwMRgwFgYD
-VQQDEw9FWEFNUExFIENPUlAgQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEK
-AoIBAQDVBr2XF20SKmGXfvp2BUY44BF1fm3Qw83GkC66qIHgpgQxzJ8u44P34vV1
-DYTj4XzrBpx23J0nCMEiAW8YW9MxZJPEokZafRNKFQAngVdgmAW9fHemHRQfaYwe
-ySVz2UZ+z28DJ3C0QshvqXskrYstwLVs4Ss2uoX+XpeFbW9IVVg4aW+GX3LHPNTk
-7Hji13Capp6BCiS61AGQAiMWNBW2aRgHVLOb7RDQug2Bka/dcJ44XxJDh26C6kV1
-pNEZm0ToqWAiVfvyzionPec5im6NKX9kStZc+C/iikfx5vdmrvGIF5zxy6phdhJ7
-AGd6cF6UBkkYm/SsIP3WTXpBek5LAgMBAAGjQjBAMA4GA1UdDwEB/wQEAwIBBjAP
-BgNVHRMBAf8EBTADAQH/MB0GA1UdDgQWBBSPLNlCmmp9PH+crzCx9TNp5BoCajAN
-BgkqhkiG9w0BAQsFAAOCAQEAkjLnZKqLMjbu5Hy8YOVoKaEKahP+t/tNYZW/NtrC
-xSxKmGLz5kVCkjpLgCykkcpu4h2wZW4MGZJ4nc87+tN22mJOg1WnqEpBzRsEPajk
-7duVjdU2FDCarERj53hxF41l/u8BxvMGbFnrAijv5SzspeLi+VCGF3wu53Bi6UHi
-7WO1dIhZLMHepqgtwbDDudxwlKbG/JWjJ7OspCNqOpktpY8BFUAQ42ctYZcOHy2w
-kdZDcK5IhRztkBH2aSWOhudYh/CF7BOu2nSUdtR5NwhXnsStmSbud7ahw3W5Jc9/
-IVAFXf2brkrfIzwrY5kmjnEPJ8mjUyLepwSlOlMTBrehzQ==
------END CERTIFICATE-----

balancer/src/test/resources/ssl/testServer.cert

@@ -1,51 +0,0 @@
-Bag Attributes
-    localKeyID: 22 89 1C EA 4E FF 9A F6 11 1B E7 9B 5E 4F 59 B7 B1 0F AA FA 
-    friendlyName: example-server
-subject=/C=US/O=Example Corp/OU=Servers/CN=testserver.example.com
-issuer=/C=US/O=Example Corp/CN=EXAMPLE CORP CA
------BEGIN CERTIFICATE-----
-MIIDwjCCAqqgAwIBAgIUe6yBgkY1m/oXuPXA9j/c/7+w16UwDQYJKoZIhvcNAQEL
-BQAwPjELMAkGA1UEBhMCVVMxFTATBgNVBAoTDEV4YW1wbGUgQ29ycDEYMBYGA1UE
-AxMPRVhBTVBMRSBDT1JQIENBMCAXDTIzMDIwMTIzMTUwMFoYDzIxMjMwMTA4MjMx
-NTAwWjBXMQswCQYDVQQGEwJVUzEVMBMGA1UEChMMRXhhbXBsZSBDb3JwMRAwDgYD
-VQQLEwdTZXJ2ZXJzMR8wHQYDVQQDExZ0ZXN0c2VydmVyLmV4YW1wbGUuY29tMIIB
-IjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAmb1pEWg9a6ZChQULazK75ppc
-XScU5Ap29L70/5zcaOcbaBwqKhpKhdZCoZ1TD+Sdkk9kp7IlabCQjM2vVjknb6NY
-msl3PbQFNva0EkH6s5cvpX4WhQDsWK77/4eztHT+0fpaG1YoKQn+Gy3VuW0CDyR6
-GzGKifnI/aNLevgUBebbiiTPLEdGlo0jp3SXlZfoDZdbvnhBWI8Xc+aDekNlrXxW
-X2TVnc02Fi914PECYJzkMpmxnBRPAcljU24WSqaUVgRoSbafkm4OXDw0xtvxnIrJ
-s3iO0uZfRf7+OxAlqlvh4ZJ9/PufPMEKvmr6zea05KkFjBotKMip5Md43H3lQQID
-AQABo4GcMIGZMA4GA1UdDwEB/wQEAwIFoDAdBgNVHSUEFjAUBggrBgEFBQcDAQYI
-KwYBBQUHAwIwDAYDVR0TAQH/BAIwADAdBgNVHQ4EFgQUDWwMrl9/uc1wplERWxas
-bDROB4MwHwYDVR0jBBgwFoAUjyzZQppqfTx/nK8wsfUzaeQaAmowGgYDVR0RBBMw
-EYIJbG9jYWxob3N0hwR/AAABMA0GCSqGSIb3DQEBCwUAA4IBAQBfG7StgQf3WNk/
-EnaYuurwvaOOpcOV5NDcZ6tmgaXUrISPVDpyIWfsDGetNEXc+4gpefLQtGF2wuiW
-odiLc4H9JwQ/yEkVYwzrFViPdZKVvxNU2EjiwO/e/3sl+SxxkTDy6o838uTXX+X5
-Q8DabmL5L8ukcnqUHVOqW38oA1vRNBKiJX58JzOr8ezX8crZmqU9h8cNwG/V+ZXE
-5U7+hSRzLm7gRwtixj8ZsTDUiGJxOdbyUKImPClt1Se9TizB2Q/xOPTdmYB1qiiB
-+XVKkGwQPgIbHckLsZWhBor2u5cgdrBGW09Q+H88maagwq5sBHNMs33vkkgbUDpB
-6WSEBJMi
------END CERTIFICATE-----
-Bag Attributes: <No Attributes>
-subject=/C=US/O=Example Corp/CN=EXAMPLE CORP CA
-issuer=/C=US/O=Example Corp/CN=EXAMPLE CORP CA
------BEGIN CERTIFICATE-----
-MIIDTjCCAjagAwIBAgIURUj8l+ms/i9jeabIBBi19KxEhAgwDQYJKoZIhvcNAQEL
-BQAwPjELMAkGA1UEBhMCVVMxFTATBgNVBAoTDEV4YW1wbGUgQ29ycDEYMBYGA1UE
-AxMPRVhBTVBMRSBDT1JQIENBMCAXDTIzMDIwMTIzMTUwMFoYDzIxMjMwMTA4MjMx
-NTAwWjA+MQswCQYDVQQGEwJVUzEVMBMGA1UEChMMRXhhbXBsZSBDb3JwMRgwFgYD
-VQQDEw9FWEFNUExFIENPUlAgQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEK
-AoIBAQDVBr2XF20SKmGXfvp2BUY44BF1fm3Qw83GkC66qIHgpgQxzJ8u44P34vV1
-DYTj4XzrBpx23J0nCMEiAW8YW9MxZJPEokZafRNKFQAngVdgmAW9fHemHRQfaYwe
-ySVz2UZ+z28DJ3C0QshvqXskrYstwLVs4Ss2uoX+XpeFbW9IVVg4aW+GX3LHPNTk
-7Hji13Capp6BCiS61AGQAiMWNBW2aRgHVLOb7RDQug2Bka/dcJ44XxJDh26C6kV1
-pNEZm0ToqWAiVfvyzionPec5im6NKX9kStZc+C/iikfx5vdmrvGIF5zxy6phdhJ7
-AGd6cF6UBkkYm/SsIP3WTXpBek5LAgMBAAGjQjBAMA4GA1UdDwEB/wQEAwIBBjAP
-BgNVHRMBAf8EBTADAQH/MB0GA1UdDgQWBBSPLNlCmmp9PH+crzCx9TNp5BoCajAN
-BgkqhkiG9w0BAQsFAAOCAQEAkjLnZKqLMjbu5Hy8YOVoKaEKahP+t/tNYZW/NtrC
-xSxKmGLz5kVCkjpLgCykkcpu4h2wZW4MGZJ4nc87+tN22mJOg1WnqEpBzRsEPajk
-7duVjdU2FDCarERj53hxF41l/u8BxvMGbFnrAijv5SzspeLi+VCGF3wu53Bi6UHi
-7WO1dIhZLMHepqgtwbDDudxwlKbG/JWjJ7OspCNqOpktpY8BFUAQ42ctYZcOHy2w
-kdZDcK5IhRztkBH2aSWOhudYh/CF7BOu2nSUdtR5NwhXnsStmSbud7ahw3W5Jc9/
-IVAFXf2brkrfIzwrY5kmjnEPJ8mjUyLepwSlOlMTBrehzQ==
------END CERTIFICATE-----

balancer/src/test/resources/ssl/testServer.key

@@ -1,32 +0,0 @@
-Bag Attributes
-    localKeyID: 22 89 1C EA 4E FF 9A F6 11 1B E7 9B 5E 4F 59 B7 B1 0F AA FA 
-    friendlyName: example-server
-Key Attributes: <No Attributes>
------BEGIN PRIVATE KEY-----
-MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCZvWkRaD1rpkKF
-BQtrMrvmmlxdJxTkCnb0vvT/nNxo5xtoHCoqGkqF1kKhnVMP5J2ST2SnsiVpsJCM
-za9WOSdvo1iayXc9tAU29rQSQfqzly+lfhaFAOxYrvv/h7O0dP7R+lobVigpCf4b
-LdW5bQIPJHobMYqJ+cj9o0t6+BQF5tuKJM8sR0aWjSOndJeVl+gNl1u+eEFYjxdz
-5oN6Q2WtfFZfZNWdzTYWL3Xg8QJgnOQymbGcFE8ByWNTbhZKppRWBGhJtp+Sbg5c
-PDTG2/GcismzeI7S5l9F/v47ECWqW+Hhkn38+588wQq+avrN5rTkqQWMGi0oyKnk
-x3jcfeVBAgMBAAECggEAZEHC7NkfhL4NjxV1Ftkk9AMhzJWZ6ARzArQsrSysB7T+
-iGwliwNOKe1J0n74FdsubqVbrIheIYhFG61edEodT44kT/ylyBVHeUp0r5tbVBZ5
-ueB56w5pEoQ6Ta8/H6VUL7alYEJVOj4L+clwINXXi+xrvKQYBlZgHheJN43j8NxS
-9lCrVcm58rzX3i29S7ErBf7RqDabgD6j2m7rehnSq5Ps2wxwV/LXMl4Lxlwa0jz/
-LV8szNMZViPOWAqOCy7NZbwCCaCfKVUAwmNGBkE3p4JVPeLNLzCBY9G0rHVmFJLK
-yPnCB7PtCZeFTsU+8OHHWoxMFsaOO9Pp8FtB/r8PYQKBgQDI0OzEo3t/QKvbwAdz
-9Eze0LQvstzZIidTWbYMTxH0g7DIt+3p11d61GKtfeuYUTPlLodufdv/BaP8G9sI
-NzxisB3uH/787M9fwi1OzO7zMMZZCVa7Q1vZk6ta1qY3PekcYMPqjBJhxP1X3S7x
-8+mrW5vTq/l8ZRVFeLvEkh6bpQKBgQDD/MCRJoq6nzd2DoTyT9m5qO+Jlztmu9+T
-vqVOg7q0NO7U7dzyMbWgMvMFoEBidnyllU1rEj2YOo4OrI2v/JmSPUVS8VXsBkkG
-UqInV4F/g0Wp3hzrkZFMqHLa6AydugmRNkOpgurRKOiwAL0Pgvk8+bOIbKMM7JBE
-tZ6Zvi0gbQKBgD4S5r0505jYQsdgqm0bXQ+a77e20DGGLyXCm1Ves6wjcOf6gt4/
-pFjhgccjH0or/6CeK1rqKLDB65ioe6IkgNP4oKfU2BsSkF+HRD4RduNRdl5bVqX4
-AlJVJSilJSf6SS5ucuDzf1MdczpPso0z2k0PujFJ/YqILHUXBuOST0fJAoGAZAdX
-nX3HZpKmsumcLzRUEq9HerkmRzvyU60FH/qZyyoND75Le42lfpsicCenVNnWqoh2
-WASaTVztbk3hjwXkt7/v5vJoASwAD2sTefLulE9Cuf00U4MbXr3GyTWQYtw1z9ib
-ViQMhy5S/KmT1Rvq65UibanM2mbPB1KPHTb0P00CgYEAgieJ1YXYYHPjABWJ7/S9
-3Z/8eJY0kuUe1DRFaKOlfej8SOCFFIa90IW2mAnujh55H4N2lww4s/N+FIRQoQpA
-1Hrch9tYRiXyouQc67dFlLxmCOVJIf31ZUt7I4HcPhsfr4fuCJXUPBs1jaDmwQsk
-90VWrXaTwSjmP+yHMiPfaA0=
------END PRIVATE KEY-----

balancer/src/test/resources/ssl/testUser.cert

@@ -1,50 +0,0 @@
-Bag Attributes
-    localKeyID: 57 F6 9F DF B2 94 61 47 B0 B0 2E 3A F7 79 98 C8 15 5C 64 FA 
-    friendlyName: example-client
-subject=/C=US/O=Example Corp/OU=Example Developers/CN=Test A. User
-issuer=/C=US/O=Example Corp/CN=EXAMPLE CORP CA
------BEGIN CERTIFICATE-----
-MIIDmzCCAoOgAwIBAgIUFp4IuIaZ2XId7UNvzP6ZB9hChyUwDQYJKoZIhvcNAQEL
-BQAwPjELMAkGA1UEBhMCVVMxFTATBgNVBAoTDEV4YW1wbGUgQ29ycDEYMBYGA1UE
-AxMPRVhBTVBMRSBDT1JQIENBMCAXDTIzMDIwMTIzMTUwMFoYDzIxMjMwMTA4MjMx
-NTAwWjBYMQswCQYDVQQGEwJVUzEVMBMGA1UEChMMRXhhbXBsZSBDb3JwMRswGQYD
-VQQLExJFeGFtcGxlIERldmVsb3BlcnMxFTATBgNVBAMTDFRlc3QgQS4gVXNlcjCC
-ASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAMOMeN0ElMDxYKST2VD1wP6Z
-jmy543Zv/QgRLTYizg3U9t3uZU0C3HqUJnK+c4+OJnRIyHuMXuJWEYVdairyAdYW
-ZDF5sfiL9ikG9rZ2WxR2z7nsYLJcDGwKAt/Duzuoed6KSccEpiQc+BkoOTs9p87I
-l4s7saX/anESMUY7YOomLSqS2zvQZh/zTjIdP1/9kG9PLzXPClJELuiC3pJCmr7Y
-xdJIXecu6YKFpApy0y+rKcfYUVPML15gyRu9pIqH3U39y7mCkn+YP9Qc9GcZAoqR
-iYdUEl1rFAgV/a3o0pmf1tFFRDOsoPSMUGNNQlaCjv1uZIbRm25nvvQlPfstFDEC
-AwEAAaN1MHMwDgYDVR0PAQH/BAQDAgWgMBMGA1UdJQQMMAoGCCsGAQUFBwMCMAwG
-A1UdEwEB/wQCMAAwHQYDVR0OBBYEFPAe3n59DcNwRcnpIwQY2kPawbCVMB8GA1Ud
-IwQYMBaAFI8s2UKaan08f5yvMLH1M2nkGgJqMA0GCSqGSIb3DQEBCwUAA4IBAQBo
-+zadU9DwR1rnHTBZF6CJXmiIdrQmArojH4a3NOHbNjFjRXm7dkd/tAy5LtgDJkww
-yg+ZrOyOKHbFgk6IXRZ8nw2b1Ck9I9MDW1vxfjid9YLW2nhDaMcIeTXz3aIBO5i6
-gMCV+xMcFaU/O57Ohia+ADUqrqu14nG9Moi171UWPbMr5N3gYn3Fs4VrTGYs4i0Z
-5x2itm88WB3bWj+cd06zKQopU9b0Y76mQv9h+w4EHxq4uWA7GmIDDkvtI61MQo8+
-twdX159Ms6o4lW43SRqVBHeMgSJbiLDkdkypYJvd9PvAjWgmQK/sD1SYKxfgYhtx
-hqIvAjlGeDwdf+55EGjf
------END CERTIFICATE-----
-Bag Attributes: <No Attributes>
-subject=/C=US/O=Example Corp/CN=EXAMPLE CORP CA
-issuer=/C=US/O=Example Corp/CN=EXAMPLE CORP CA
------BEGIN CERTIFICATE-----
-MIIDTjCCAjagAwIBAgIURUj8l+ms/i9jeabIBBi19KxEhAgwDQYJKoZIhvcNAQEL
-BQAwPjELMAkGA1UEBhMCVVMxFTATBgNVBAoTDEV4YW1wbGUgQ29ycDEYMBYGA1UE
-AxMPRVhBTVBMRSBDT1JQIENBMCAXDTIzMDIwMTIzMTUwMFoYDzIxMjMwMTA4MjMx
-NTAwWjA+MQswCQYDVQQGEwJVUzEVMBMGA1UEChMMRXhhbXBsZSBDb3JwMRgwFgYD
-VQQDEw9FWEFNUExFIENPUlAgQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEK
-AoIBAQDVBr2XF20SKmGXfvp2BUY44BF1fm3Qw83GkC66qIHgpgQxzJ8u44P34vV1
-DYTj4XzrBpx23J0nCMEiAW8YW9MxZJPEokZafRNKFQAngVdgmAW9fHemHRQfaYwe
-ySVz2UZ+z28DJ3C0QshvqXskrYstwLVs4Ss2uoX+XpeFbW9IVVg4aW+GX3LHPNTk
-7Hji13Capp6BCiS61AGQAiMWNBW2aRgHVLOb7RDQug2Bka/dcJ44XxJDh26C6kV1
-pNEZm0ToqWAiVfvyzionPec5im6NKX9kStZc+C/iikfx5vdmrvGIF5zxy6phdhJ7
-AGd6cF6UBkkYm/SsIP3WTXpBek5LAgMBAAGjQjBAMA4GA1UdDwEB/wQEAwIBBjAP
-BgNVHRMBAf8EBTADAQH/MB0GA1UdDgQWBBSPLNlCmmp9PH+crzCx9TNp5BoCajAN
-BgkqhkiG9w0BAQsFAAOCAQEAkjLnZKqLMjbu5Hy8YOVoKaEKahP+t/tNYZW/NtrC
-xSxKmGLz5kVCkjpLgCykkcpu4h2wZW4MGZJ4nc87+tN22mJOg1WnqEpBzRsEPajk
-7duVjdU2FDCarERj53hxF41l/u8BxvMGbFnrAijv5SzspeLi+VCGF3wu53Bi6UHi
-7WO1dIhZLMHepqgtwbDDudxwlKbG/JWjJ7OspCNqOpktpY8BFUAQ42ctYZcOHy2w
-kdZDcK5IhRztkBH2aSWOhudYh/CF7BOu2nSUdtR5NwhXnsStmSbud7ahw3W5Jc9/
-IVAFXf2brkrfIzwrY5kmjnEPJ8mjUyLepwSlOlMTBrehzQ==
------END CERTIFICATE-----

balancer/src/test/resources/ssl/testUser.key

@@ -1,32 +0,0 @@
-Bag Attributes
-    localKeyID: 57 F6 9F DF B2 94 61 47 B0 B0 2E 3A F7 79 98 C8 15 5C 64 FA 
-    friendlyName: example-client
-Key Attributes: <No Attributes>
------BEGIN PRIVATE KEY-----
-MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDDjHjdBJTA8WCk
-k9lQ9cD+mY5sueN2b/0IES02Is4N1Pbd7mVNAtx6lCZyvnOPjiZ0SMh7jF7iVhGF
-XWoq8gHWFmQxebH4i/YpBva2dlsUds+57GCyXAxsCgLfw7s7qHneiknHBKYkHPgZ
-KDk7PafOyJeLO7Gl/2pxEjFGO2DqJi0qkts70GYf804yHT9f/ZBvTy81zwpSRC7o
-gt6SQpq+2MXSSF3nLumChaQKctMvqynH2FFTzC9eYMkbvaSKh91N/cu5gpJ/mD/U
-HPRnGQKKkYmHVBJdaxQIFf2t6NKZn9bRRUQzrKD0jFBjTUJWgo79bmSG0ZtuZ770
-JT37LRQxAgMBAAECggEBALFWsT1vLCAWRxi55Hkhhhbe12UO55O/V+969E+GHOw6
-WG9IBaJjl+Oqe3vP8CkeNw15/CbfVCVa/KS/5Mpxkuy+fbrIrHfFJkZqxE8Q4uSa
-IyRvk/nstG8+XQDlFuWxkmQZcXMFak6R3wWhjtWEwqwrChQA3Mi366YDY6BXtTpi
-MuLJVWXqYVF2RirWX3a7eosR4MYBCJ1S8NnpI1w4oZwuew6iYiCJNkURctxnGKOe
-hoop/711PEyIb/qHETBIlReY9scvZgJQZMgGe5mTGgFmAA8Bzo4e1/F2QrVIi7wy
-REOSj4U+tP/QrYMSPk0cMpH5ft5+2mfh/o+D1ugm/AkCgYEA2qt3P4vlm1TMmJru
-vMYc8iZjfe7UQTnD+J1EgrASBI0JP565zSZKBRIXnheumWhWS+2KR1MP8N6KAu4Q
-xE/E7TKfLuE1CPzn7Rh/WZAxDqGNeQthk0qTbeBZ4/mHPKvt8V7c+N4vLjZ00uul
-bKitlLpb49T+8FpSgRmk+nrBsesCgYEA5O6LmxH1+ip5NoMHoZVBhLbgJkxoXEzn
-xHT7On8CjQ9B0T85NW9O38/a2gyIb5dIavcrh4dMvOKX8hdrVlal3chrXFRemyvy
-VzpW7HZpMwthJb/QSim+GQ4u2Jsm+eXQAswq/yfowls9mp7K81McBgdfcdWgWuqc
-Ttf3dBRS71MCgYBCPpu7ppgCfKgWuykEC0iNm0+cwROQRv21fX9Z0BlFQ6y/l0hW
-e/kEnDyHakk8tD0gd2/it6kmoWRSCukNb9UKvsE2pIc6l9NzIm7CJUqXUh3Xw7jl
-tDwBl2vkAp4TpUpKSyDKTBUi/OBbXYQZ7OfZHLPly4sIKXTBI5G1SCcRzQKBgE3Y
-Y7ZmRBs+CBumMJkcVLwZJ10+jBoeuIzQLoF0oNxwfLP3aK+n2s02M7gDnP09Fw3w
-YGfkdQ6hioNyvTgNnLcuqHsr4UOojO5mUxpzJVCLJQyeDei6Cv8gGIPlNjqxj6rz
-3AcJfhdgyTDjoCVfdF8EcMsXlkB0GJSi0W0yYAUrAoGAVK5v0cbEEV+gxdQgh848
-FwSQXx7i9jVOB2BUTLX2Q+gLUJibtFzLnh03sGCXtTa8sK+hLW4FCMWZa9Ev5ZjG
-ylFyA96U97CNpS+2/ZW5+hvRZ/GuY/0yB8w26BeS38lI8RnmxSPi5cnexdl32NIX
-5FkU9WJ4HKmi1n0304CkF+4=
------END PRIVATE KEY-----

client/pom.xml

@@ -4,13 +4,12 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.6-SNAPSHOT</version>
+    <version>1.0.1-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.6.dev0</python.client.version>
+    <python.client.version>0.0.7.dev0</python.client.version>
   </properties>
   <dependencies>
     <dependency>
@@ -53,10 +52,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 +62,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>
@@ -109,43 +107,4 @@
       </plugin>
     </plugins>
   </build>
-  <profiles>
-    <profile>
-      <id>python</id>
-      <build>
-        <plugins>
-          <plugin>
-            <groupId>org.codehaus.mojo</groupId>
-            <artifactId>exec-maven-plugin</artifactId>
-            <executions>
-              <execution>
-                <id>python-dist</id>
-                <goals>
-                  <goal>exec</goal>
-                </goals>
-                <phase>package</phase>
-                <configuration>
-                  <executable>python</executable>
-                  <workingDirectory>${project.basedir}/src/main/python</workingDirectory>
-                  <commandlineArgs>setup.py sdist --dist-dir ${project.build.directory}</commandlineArgs>
-                </configuration>
-              </execution>
-              <execution>
-                <id>python-clean</id>
-                <goals>
-                  <goal>exec</goal>
-                </goals>
-                <phase>package</phase>
-                <configuration>
-                  <executable>rm</executable>
-                  <workingDirectory>${project.basedir}/src/main/python</workingDirectory>
-                  <commandlineArgs>-rf timely.egg-info setup.py</commandlineArgs>
-                </configuration>
-              </execution>
-            </executions>
-          </plugin>
-        </plugins>
-      </build>
-    </profile>
-  </profiles>
 </project>

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

@@ -1,13 +1,12 @@
 package timely.client.http;
 
-import javax.net.ssl.HostnameVerifier;
 import javax.net.ssl.SSLContext;
 
 import org.apache.http.client.CookieStore;
 import org.apache.http.client.config.CookieSpecs;
 import org.apache.http.client.config.RequestConfig;
+import org.apache.http.conn.ssl.DefaultHostnameVerifier;
 import org.apache.http.conn.ssl.NoopHostnameVerifier;
-import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.client.HttpClientBuilder;
 import org.apache.http.impl.client.HttpClients;
@@ -15,8 +14,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,37 +30,24 @@ 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);
     }
 
     public static CloseableHttpClient get(SSLContext ssl, CookieStore cookieStore, boolean hostVerificationEnabled) {
-        return get(ssl, cookieStore, hostVerificationEnabled, false);
-    }
-
-    public static CloseableHttpClient get(SSLContext ssl, CookieStore cookieStore, boolean hostVerificationEnabled, boolean clientAuth) {
-        HostnameVerifier hostnameVerifier;
-        if (hostVerificationEnabled) {
-            hostnameVerifier = SSLConnectionSocketFactory.getDefaultHostnameVerifier();
-        } else {
-            hostnameVerifier = new NoopHostnameVerifier();
-        }
         RequestConfig defaultRequestConfig = RequestConfig.custom().setCookieSpec(CookieSpecs.STANDARD).build();
-        HttpClientBuilder builder;
-        if (cookieStore == null) {
-            builder = HttpClients.custom().setSSLContext(ssl).disableCookieManagement().setDefaultRequestConfig(defaultRequestConfig)
-                            .setSSLHostnameVerifier(hostnameVerifier);
-        } else {
-            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);
-            builder.setSSLSocketFactory(sslCSF);
+        HttpClientBuilder builder = HttpClients.custom().setSSLContext(ssl).setDefaultCookieStore(cookieStore)
+                .setDefaultRequestConfig(defaultRequestConfig);
+        if (hostVerificationEnabled) {
+            builder.setSSLHostnameVerifier(new DefaultHostnameVerifier());
+        } else {
+            builder.setSSLHostnameVerifier(new NoopHostnameVerifier());
         }
         return builder.build();
     }

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

@@ -1,22 +1,20 @@
 package timely.client.tcp;
 
-import java.io.BufferedWriter;
 import java.io.IOException;
-import java.io.OutputStreamWriter;
+import java.io.PrintWriter;
 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;
     private Socket sock = null;
-    private BufferedWriter out = null;
+    private PrintWriter out = null;
     private long connectTime = 0L;
     private long backoff = 2000;
     private int bufferSize = -1;
@@ -34,7 +32,7 @@ public class TcpClient implements AutoCloseable {
 
     /**
      * Opens a TCP connection to the specified host and port
-     *
+     * 
      * @throws IOException
      *             if an error occurs
      */
@@ -46,9 +44,10 @@ public class TcpClient implements AutoCloseable {
 
     /**
      * Write a metric to Timely
-     *
+     * 
      * @param metric
      *            newline terminated string representation of Timely metric
+     * @throws IOException
      */
     public synchronized void write(String metric) throws IOException {
         if (connect() != 0) {
@@ -61,7 +60,7 @@ public class TcpClient implements AutoCloseable {
         }
     }
 
-    public synchronized void flush() throws IOException {
+    public synchronized void flush() {
         if (null != out) {
             out.flush();
             writesSinceFlush = 0;
@@ -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;
             }
@@ -95,49 +94,24 @@ public class TcpClient implements AutoCloseable {
     }
 
     private synchronized int connect() {
-        if (null == sock || !sock.isConnected()) {
+        if (null == sock || !sock.isConnected() || out.checkError()) {
             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 PrintWriter(sock.getOutputStream(), false);
                     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

@@ -1,11 +1,9 @@
 package timely.client.websocket;
 
-import java.util.List;
-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 +11,28 @@ 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) {}
 }

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

@@ -14,15 +14,13 @@ import org.apache.http.cookie.Cookie;
 public class TimelyEndpointConfig implements ClientEndpointConfig {
 
     private String sessionCookie = null;
-    private final ClientHandler clientEndpoint;
 
     private static final String FORMAT = "%s=%s;";
 
-    public TimelyEndpointConfig(ClientHandler clientEndpoint, Cookie sessionCookie) {
+    public TimelyEndpointConfig(Cookie sessionCookie) {
         if (null != sessionCookie) {
             this.sessionCookie = String.format(FORMAT, sessionCookie.getName(), sessionCookie.getValue());
         }
-        this.clientEndpoint = clientEndpoint;
     }
 
     @Override
@@ -36,7 +34,7 @@ public class TimelyEndpointConfig implements ClientEndpointConfig {
     }
 
     @Override
-    public Map<String,Object> getUserProperties() {
+    public Map<String, Object> getUserProperties() {
         return Collections.emptyMap();
     }
 
@@ -55,12 +53,11 @@ 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));
                 }
-                TimelyEndpointConfig.this.clientEndpoint.beforeRequest(headers);
             }
         };
     }

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,20 @@ 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,46 +50,71 @@ 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 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 && ((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 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, doLogin, timelyUsername, timelyPassword,
+                hostVerificationEnabled, bufferSize);
     }
 
     public void open(ClientHandler clientEndpoint) throws IOException, DeploymentException, URISyntaxException {
+
         Cookie sessionCookie = null;
         if (doLogin) {
             BasicCookieStore cookieJar = new BasicCookieStore();
-            try (CloseableHttpClient client = HttpClient.get(ssl, cookieJar, hostVerificationEnabled, clientAuth)) {
+            try (CloseableHttpClient client = HttpClient.get(ssl, cookieJar, hostVerificationEnabled)) {
 
                 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 +123,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 +140,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(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 doLogin, String timelyUsername, String timelyPassword, boolean hostVerificationEnabled,
+            int bufferSize) {
+        super(ssl, timelyHostname, timelyHttpsPort, timelyWssPort, 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 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, 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,18 @@ 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.
          *
+         * @param o
          * @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 +76,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 +88,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 +159,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,16 +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 org.apache.commons.lang3.builder.HashCodeBuilder;
-
 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;
 
 /**
@@ -30,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());
@@ -49,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));
     }
 
@@ -79,6 +77,8 @@ public class Tag implements Comparable<Tag>, Serializable {
 
     /**
      * Joins key and value into key=value
+     * 
+     * @return
      */
     public String join() {
         return tagParser.combine(this);
@@ -91,7 +91,7 @@ public class Tag implements Comparable<Tag>, Serializable {
 
     @Override
     public int hashCode() {
-        return new HashCodeBuilder().append(this.key).append(this.value).toHashCode();
+        return this.toString().hashCode();
     }
 
     @Override

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

@@ -18,7 +18,8 @@ public class Value implements Comparator<Value>, Serializable {
     private Long timestamp;
     private Double measure;
 
-    public Value() {}
+    public Value() {
+    }
 
     public Value(Value other) {
         this.setTimestamp(other.getTimestamp());
@@ -73,9 +74,11 @@ public class Value implements Comparator<Value>, Serializable {
         if (null == o || getClass() != o.getClass())
             return false;
 
-        if (getTimestamp() != null ? !getTimestamp().equals(((Value) o).getTimestamp()) : ((Value) o).getTimestamp() != null)
+        if (getTimestamp() != null ? !getTimestamp().equals(((Value) o).getTimestamp())
+                : ((Value) o).getTimestamp() != null)
             return false;
-        return !(getMeasure() != null ? !getMeasure().equals(((Value) o).getMeasure()) : ((Value) o).getMeasure() != null);
+        return !(getMeasure() != null ? !getMeasure().equals(((Value) o).getMeasure())
+                : ((Value) o).getMeasure() != null);
     }
 
     @Override