Release version 2.0.2

LICENSE

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

README.md

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

analytics/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>2.0.2</version>
   </parent>
   <artifactId>timely-analytics</artifactId>
   <!--
@@ -53,15 +53,16 @@
         </exclusion>
       </exclusions>
     </dependency>
-    <dependency>
-      <groupId>timely</groupId>
-      <artifactId>timely-common</artifactId>
-    </dependency>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>timely</groupId>
+      <artifactId>timely-common</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
   <build>
     <pluginManagement>

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

@@ -42,7 +42,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 +64,11 @@ 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());
+                        jp.getTimelyUsername(), jp.getTimelyPassword(), jp.getKeyStoreFile(), jp.getKeyStoreType(), jp.getKeyStorePass(),
+                        jp.getTrustStoreFile(), jp.getTrustStoreType(), jp.getTrustStorePass(), jp.isHostVerificationEnabled(), jp.getBufferSize());
     }
 
     @Override
@@ -81,7 +81,7 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
 
     @Override
     public void run(SourceContext<MetricResponse> ctx) throws Exception {
-        log.info("Running summarization job");
+        LOG.info("Running summarization job");
         final SortedStringAccumulator dateTimeAccumulator = new SortedStringAccumulator();
         this.getRuntimeContext().addAccumulator("source hourly count", dateTimeAccumulator);
         final LongCounter sourceInputs = new LongCounter();
@@ -103,7 +103,7 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
 
                     @Override
                     public void onMessage(final String message) {
-                        log.info("Message received on Websocket session {}, length: {}", session.getId(), message.length());
+                        LOG.info("Message received on Websocket session {}, length: {}", session.getId(), message.length());
                         try {
                             // Deserialize in this thread
                             final MetricResponses responses = JsonSerializer.getObjectMapper().readValue(message, MetricResponses.class);
@@ -111,12 +111,12 @@ public class SubscriptionSource extends RichSourceFunction<MetricResponse> imple
                             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 +139,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 +156,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 +168,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 +176,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 +193,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

@@ -18,7 +18,7 @@ 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 +27,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 +43,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 +59,7 @@ public class SummarizationJob {
                         throw new RuntimeException("Tags don't not match - partitioning problem");
                     }
                 }
-                log.trace("Updating histogram for {} with value {} at {} for metric {}", histo.getMetric()
+                LOG.trace("Updating histogram for {} with value {} at {} for metric {}", histo.getMetric()
                         + histo.getTags(), metric.getValue(), metric.getTimestamp(), metric.getMetric()
                         + metric.getTags());
                 histo.update(metric.getValue(), metric.getTimestamp());
@@ -71,7 +71,7 @@ public class SummarizationJob {
             new SocketClientSink<MetricHistogram>(jp.getTimelyHostname(), jp.getTimelyTcpPort(),
                     new MetricHistogram(), 2, true) {
 
-                private final Logger log = LoggerFactory
+                private final Logger LOG = LoggerFactory
                         .getLogger("timely.analytics.SummarizationJob.SocketClientSink");
                 private static final long serialVersionUID = 1L;
                 private final LongCounter sinkCounter = new LongCounter();
@@ -83,7 +83,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 +97,7 @@ public class SummarizationJob {
                 @Override
                 public void close() throws Exception {
                     super.close();
-                    log.info("Sink closed");
+                    LOG.info("Sink closed");
                 }
 
             }).name("Timely Metric Sink");

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

@@ -22,11 +22,7 @@ for i in $(seq 1 $NUM_PROFILES); do
 
     echo "creating configuration file ${CONF_DIR}/${FILENAME}"
     cp ${CONF_DIR}/timely-balancer-template.yml ${CONF_DIR}/${FILENAME}
-#    if [ "$NUM_PROFILES" -eq "1" ]; then
-#        sed -i "s/\${INSTANCE}//g" ${CONF_DIR}/${FILENAME}
-#    else
-#        sed -i "s/\${INSTANCE}/${i}/g" ${CONF_DIR}/${FILENAME}
-#    fi
+    sed -i "s/\${ZOOKEEPERS}/${ZOOKEEPERS}/g" ${CONF_DIR}/${FILENAME}
     MULTIPLE=$(($i - 1))
     TCP_PORT=$(($TCP_PORT_START + $MULTIPLE * $PORT_INCREMENT))
     sed -i "s/\${TCP_PORT}/${TCP_PORT}/g" ${CONF_DIR}/${FILENAME}
@@ -36,5 +32,16 @@ for i in $(seq 1 $NUM_PROFILES); do
     sed -i "s/\${WEBSOCKET_PORT}/${WEBSOCKET_PORT}/g" ${CONF_DIR}/${FILENAME}
     UDP_PORT=$(($UDP_PORT_START + $MULTIPLE * $PORT_INCREMENT))
     sed -i "s/\${UDP_PORT}/${UDP_PORT}/g" ${CONF_DIR}/${FILENAME}
+    sed -i "s/\${INSTANCE}/${i}/g" ${CONF_DIR}/${FILENAME}
+    sed -i "s#\${SERVER_CERTIFICATE_FILE}#${SERVER_CERTIFICATE_FILE}#g" ${CONF_DIR}/${FILENAME}
+    sed -i "s#\${SERVER_KEY_FILE}#${SERVER_KEY_FILE}#g" ${CONF_DIR}/${FILENAME}
+    sed -i "s/\${SERVER_KEY_PASSWORD}/${SERVER_KEY_PASSWORD}/g" ${CONF_DIR}/${FILENAME}
+    sed -i "s/\${SERVER_USE_GENERATED_KEYPAIR}/${SERVER_USE_GENERATED_KEYPAIR}/g" ${CONF_DIR}/${FILENAME}
+    sed -i "s#\${CLIENT_KEY_FILE}#${CLIENT_KEY_FILE}#g" ${CONF_DIR}/${FILENAME}
+    sed -i "s/\${CLIENT_KEY_TYPE}/${CLIENT_KEY_TYPE}/g" ${CONF_DIR}/${FILENAME}
+    sed -i "s/\${CLIENT_KEY_PASSWORD}/${CLIENT_KEY_PASSWORD}/g" ${CONF_DIR}/${FILENAME}
+    sed -i "s#\${TRUST_STORE_PEM}#${TRUST_STORE_PEM}#g" ${CONF_DIR}/${FILENAME}
+    sed -i "s#\${TRUST_STORE_JKS}#${TRUST_STORE_JKS}#g" ${CONF_DIR}/${FILENAME}
+    sed -i "s/\${TRUST_STORE_PASSWORD}/${TRUST_STORE_PASSWORD}/g" ${CONF_DIR}/${FILENAME}
 done
 

balancer/bin/stop-all.sh

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

balancer/bin/timely-balancer-env.sh

@@ -1,7 +1,7 @@
 #!/bin/bash
 
 NUM_PROFILES=1
-ACCUMULO_ZOOKEEPERS=localhost:2181
+ZOOKEEPERS=localhost:2181
 TCP_PORT_START=4242
 HTTP_PORT_START=4243
 WEBSOCKET_PORT_START=4244
@@ -22,6 +22,3 @@ KEY_STORE_PASSWORD=
 AUTHORIZATION_URL=
 AUTHORIZATION_CERTIFICATE_TYPE=
 AUTHORIZATION_CERTIFICATE_FILE=
-
-REQUIRED_ROLES=
-REQUIRED_AUTHS=

balancer/bin/timely-balancer.sh

@@ -15,12 +15,12 @@ set -a
 set +a
 
 if [ -n "$1" ]; then
-  export INSTANCE="$1"
-  PROFILE_ARG="--spring.profiles.active=${INSTANCE}"
-  PID=`jps -m | grep -E 'timely-balancer-.*-exec' | grep "spring.profiles.active=${INSTANCE}" | awk '{print $1}'`
+  export PROFILE_NUM=$1
+  PROFILE_ARG="--spring.profiles.active=${PROFILE_NUM}"
+  PID=`jps -m | grep timely.balancer.Balancer | grep "spring.profiles.active=${PROFILE_NUM}" | awk '{print $1}'`
 else
-  export INSTANCE="1"
-  PID=`jps -m | grep -E 'timely-balancer-.*-exec' | awk '{print $1}'`
+  export PROFILE_NUM=""
+  PID=`jps -m | grep timely.balancer.Balancer | awk '{print $1}'`
 fi
 
 if [ "$PID" == "" ]; then
@@ -28,7 +28,7 @@ if [ "$PID" == "" ]; then
     NATIVE_DIR="${THIS_DIR}/META-INF/native"
     BASE_DIR=$(cd $THIS_DIR/.. && pwd)
     TMP_DIR="${BASE_DIR}/tmp"
-    export CONF_DIR="${BASE_DIR}/conf"
+    CONF_DIR="${BASE_DIR}/conf"
     LIB_DIR="${BASE_DIR}/lib"
     export LOG_DIR="${BASE_DIR}/logs"
     NUM_SERVER_THREADS=4
@@ -53,7 +53,7 @@ if [ "$PID" == "" ]; then
     popd
 
     export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*:${HADOOP_CONF_DIR}"
-    JVM_ARGS="-Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} -Dlog4j.configurationFile=${CONF_DIR}/log4j2.yml"
+    JVM_ARGS="-Xmx1G -Xms1G -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} -Dlog4j.configurationFile=${CONF_DIR}/log4j2-spring.xml"
     JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
     JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
 
@@ -61,9 +61,13 @@ if [ "$PID" == "" ]; then
     JVM_ARGS="${JVM_ARGS} -XX:+UseCompressedOops -XX:+UseNUMA -Djava.net.preferIPv4Stack=true"
     JVM_ARGS="${JVM_ARGS} -XX:+UseG1GC -XX:MaxGCPauseMillis=4000 -XX:ParallelGCThreads=20 -XX:+UseStringDeduplication"
 
-    echo "$JAVA_HOME/bin/java ${JVM_ARGS} -jar ${THIS_DIR}/timely-balancer-*-exec.jar ${PROFILE_ARG}"
-    nohup $JAVA_HOME/bin/java ${JVM_ARGS} -jar ${THIS_DIR}/timely-balancer-*-exec.jar ${PROFILE_ARG} >> ${LOG_DIR}/timely-balancer${INSTANCE}.out 2>&1 &
+    echo "$JAVA_HOME/bin/java ${JVM_ARGS} timely.balancer.Balancer --spring.config.name=timely-balancer ${PROFILE_ARG}"
+    nohup $JAVA_HOME/bin/java ${JVM_ARGS} timely.balancer.Balancer --spring.config.name=timely-balancer ${PROFILE_ARG} >> ${LOG_DIR}/timely-balancer${PROFILE_NUM}-startup.log 2>&1 &
 
 else
-    echo "timely-balancer ${INSTANCE} already running with pid ${PID}"
+    if [ "$PROFILE_NUM" == ""]; then
+        echo "timely-balancer already running with pid ${PID}"
+    else
+        echo "timely-balancer ${PROFILE_NUM} already running with pid ${PID}"
+    fi
 fi

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,23 @@
+<?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="${env:LOG_DIR}/timely-balancer${env:PROFILE_NUM}.log"
+    						 filePattern="${env:LOG_DIR}/$${date:yyyy-MM}/timely-balancer${env:PROFILE_NUM}-%d{MM-dd-yyyy}-%i.log.gz">
+      <PatternLayout pattern="${LOG_PATTERN}"/>
+      <Policies>
+      	<SizeBasedTriggeringPolicy size="1024 MB" />
+      </Policies>
+      <DefaultRolloverStrategy max="10" />
+    </RollingRandomAccessFile>
+  </Appenders>
+  <Loggers>
+    <Root level="warn" includeLocation="false">
+      <AppenderRef ref="Rolling"/>
+    </Root>
+    <Logger name="timely" level="debug"/>
+  </Loggers>
+</Configuration>

balancer/conf/log4j2.yml

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

balancer/conf/security.xml

@@ -0,0 +1,64 @@
+<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"
+			 xmlns:context="http://www.springframework.org/schema/context"
+			 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 http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
+
+	<context:property-placeholder location="classpath:security.properties"/>
+
+	<beans:bean id="clientSsl" class="timely.configuration.ClientSsl">
+		<beans:property name="trustStoreType" value="${trust.store.type}"/>
+		<beans:property name="trustStoreFile" value="${trust.store.file}"/>
+		<beans:property name="trustStorePassword" value="${trust.store.password}"/>
+		<beans:property name="keyStoreType" value="${key.store.type}"/>
+		<beans:property name="keyStoreFile" value="${key.store.file}"/>
+		<beans:property name="keyStorePassword" value="${key.store.password}"/>
+	</beans:bean>
+
+	<beans:bean id="authorizationProperties" class="timely.configuration.AuthorizationProperties">
+		<beans:property name="authorizationUrl" value="${authorization.url}"/>
+		<beans:property name="publicCertificateType" value="${authorization.certificate.type}"/>
+		<beans:property name="publicCertificateFile" value="${authorization.certificate.file}"/>
+	</beans:bean>
+
+	<beans:bean id="remoteUserDetailsService" class="timely.auth.RemoteUserDetailsService">
+		<beans:constructor-arg ref="clientSsl"/>
+		<beans:constructor-arg ref="authorizationProperties"/>
+	</beans:bean>
+
+	<beans:bean id="x509AuthorizationProvider" class="org.springframework.security.web.authentication.preauth.PreAuthenticatedAuthenticationProvider">
+		<beans:property name="preAuthenticatedUserDetailsService" ref="remoteUserDetailsService" />
+	</beans:bean>
+
+	<user-service id="usernamePasswordService">
+		<user name="test" password="test1" authorities="A,B,C" />
+	</user-service>
+
+	<beans:bean id="noOpPasswordEncoder" class="org.springframework.security.crypto.password.NoOpPasswordEncoder"/>
+
+	<beans:bean id="UserNamePasswordAuthenticationProvider" class="org.springframework.security.authentication.dao.DaoAuthenticationProvider">
+		<beans:property name="userDetailsService" ref="usernamePasswordService"/>
+		<beans:property name="passwordEncoder" ref="noOpPasswordEncoder"/>
+	</beans:bean>
+
+	<authentication-manager alias="authenticationManager">
+		<!-- X509 PreAuthenticated authorization provider -->
+		<authentication-provider ref="x509AuthorizationProvider"/>
+		<!-- Basic authorization provider, validates provided username and password -->
+		<authentication-provider ref="UserNamePasswordAuthenticationProvider" />
+	</authentication-manager>
+
+	<util:list id="requiredRoles" value-type="java.lang.String">
+		<beans:value></beans:value>
+	</util:list>
+
+	<util:list id="requiredAuths" value-type="java.lang.String">
+		<beans:value></beans:value>
+	</util:list>
+</beans:beans>

balancer/conf/timely-balancer-template.yml

@@ -1,7 +1,11 @@
-timely:
-  zookeeper:
-    servers: ${ACCUMULO_ZOOKEEPERS}
+timely-balancer:
+  zooKeeper:
+    servers: ${ZOOKEEPERS}
     timeout: 30s
+  checkServerHealthInterval: 10000
+  serverFailuresBeforeDown: 6
+  serverSuccessesBeforeUp: 1
+  assignmentFile: /timely/balancer/assignments.csv
   security:
     session-max-age: 86400
     cache-expiration-minutes: 60
@@ -33,110 +37,100 @@ timely:
     metric-age-off-hours:
       default: 26
     nonCachedMetrics:
-  balancer:
-    checkServerHealthInterval: 10000
-    serverFailuresBeforeDown: 6
-    serverSuccessesBeforeUp: 1
-    assignmentFile: /timely/balancer/assignments.csv
-    server:
-      ip: ${HOST_IP}
-      tcpPort: ${TCP_PORT}
-      numTcpPools: 1
-      tcpBufferSize: 100
-      tcpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-      udpPort: ${UDP_PORT}
-      udpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-    http:
-      host: ${HOST}
-      ip: ${HOST_IP}
-      port: ${HTTP_PORT}
-      redirect-path: /secure-me
-      strict-transport-max-age: 604800
-      cors:
-        allow-any-origin: false
-        allow-null-origin: false
-        allow-credentials: true
-        allowed-origins:
-        allowed-methods: DELETE,GET,HEAD,OPTIONS,PUT,POST
-        allowed-headers: content-type
-      httpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-    websocket:
-      ip: ${HOST_IP}
-      port: ${WEBSOCKET_PORT}
-      timeout: 60
-      subscription-lag: 120
-      scanner-batch-size: 10000
-      flush-interval-seconds: 30
-      scanner-read-ahead: 1
-      subscription-batch-size: 5000
-      incomingBufferSize: 50000000
-      wsClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-
-logging:
-  level:
-    ROOT: info
-    timey: debug
+  server:
+    ip: ${HOST_IP}
+    tcpPort: ${TCP_PORT}
+    numTcpPools: 1
+    tcpBufferSize: 100
+    tcpClientPool:
+      minIdlePerKey: 0
+      maxIdlePerKey: 40
+      maxTotalPerKey: 40
+      maxTotal: -1
+      lifo: true
+      fairness: false
+      maxWaitMillis: -1
+      minEvictableIdleTimeMillis: 1800000
+      evictorShutdownTimeoutMillis: 10000
+      softMinEvictableIdleTimeMillis: -1
+      numTestsPerEvictionRun: 3
+      timeBetweenEvictionRunsMillis: -1
+      blockWhenExhausted: true
+      jmxEnabled: true
+      testOnBorrow: false
+      testOnReturn: false
+    udpPort: ${UDP_PORT}
+    udpClientPool:
+      minIdlePerKey: 0
+      maxIdlePerKey: 40
+      maxTotalPerKey: 40
+      maxTotal: -1
+      lifo: true
+      fairness: false
+      maxWaitMillis: -1
+      minEvictableIdleTimeMillis: 1800000
+      evictorShutdownTimeoutMillis: 10000
+      softMinEvictableIdleTimeMillis: -1
+      numTestsPerEvictionRun: 3
+      timeBetweenEvictionRunsMillis: -1
+      blockWhenExhausted: true
+      jmxEnabled: true
+      testOnBorrow: false
+      testOnReturn: false
+  http:
+    host: ${HOST}
+    ip: ${HOST_IP}
+    port: ${HTTP_PORT}
+    redirect-path: /secure-me
+    strict-transport-max-age: 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: ${HOST_IP}
+    port: ${WEBSOCKET_PORT}
+    timeout: 60
+    subscription-lag: 120
+    scanner-batch-size: 10000
+    flush-interval-seconds: 30
+    scanner-read-ahead: 1
+    subscription-batch-size: 5000
+    incomingBufferSize: 50000000
+    wsClientPool:
+      minIdlePerKey: 0
+      maxIdlePerKey: 40
+      maxTotalPerKey: 40
+      maxTotal: -1
+      lifo: true
+      fairness: false
+      maxWaitMillis: -1
+      minEvictableIdleTimeMillis: 1800000
+      evictorShutdownTimeoutMillis: 10000
+      softMinEvictableIdleTimeMillis: -1
+      numTestsPerEvictionRun: 3
+      timeBetweenEvictionRunsMillis: -1
+      blockWhenExhausted: true
+      jmxEnabled: true
+      testOnBorrow: false
+      testOnReturn: false

balancer/conf/timely-balancer.yml

@@ -1,7 +1,11 @@
-timely:
-  zookeeper:
-    servers: ${ACCUMULO_ZOOKEEPERS}
+timely-balancer:
+  zooKeeper:
+    servers: localhost:2181
     timeout: 30s
+  checkServerHealthInterval: 10000
+  serverFailuresBeforeDown: 6
+  serverSuccessesBeforeUp: 1
+  assignmentFile: /timely/balancer/assignments.csv
   security:
     session-max-age: 86400
     cache-expiration-minutes: 60
@@ -33,110 +37,100 @@ timely:
     metric-age-off-hours:
       default: 26
     nonCachedMetrics:
-  balancer:
-    checkServerHealthInterval: 10000
-    serverFailuresBeforeDown: 6
-    serverSuccessesBeforeUp: 1
-    assignmentFile: /timely/balancer/assignments.csv
-    server:
-      ip: 0.0.0.0
-      tcpPort: 4242
-      numTcpPools: 1
-      tcpBufferSize: 100
-      tcpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-      udpPort: 4245
-      udpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-    http:
-      host: localhost
-      ip: 0.0.0.0
-      port: 4243
-      redirect-path: /secure-me
-      strict-transport-max-age: 604800
-      cors:
-        allow-any-origin: false
-        allow-null-origin: false
-        allow-credentials: true
-        allowed-origins:
-        allowed-methods: DELETE,GET,HEAD,OPTIONS,PUT,POST
-        allowed-headers: content-type
-      httpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-    websocket:
-      ip: 0.0.0.0
-      port: 5244
-      timeout: 60
-      subscription-lag: 120
-      scanner-batch-size: 10000
-      flush-interval-seconds: 30
-      scanner-read-ahead: 1
-      subscription-batch-size: 5000
-      incomingBufferSize: 50000000
-      wsClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-
-logging:
-  level:
-    ROOT: info
-    timey: debug
+  server:
+    ip: 0.0.0.0
+    tcpPort: 4242
+    numTcpPools: 1
+    tcpBufferSize: 100
+    tcpClientPool:
+      minIdlePerKey: 0
+      maxIdlePerKey: 40
+      maxTotalPerKey: 40
+      maxTotal: -1
+      lifo: true
+      fairness: false
+      maxWaitMillis: -1
+      minEvictableIdleTimeMillis: 1800000
+      evictorShutdownTimeoutMillis: 10000
+      softMinEvictableIdleTimeMillis: -1
+      numTestsPerEvictionRun: 3
+      timeBetweenEvictionRunsMillis: -1
+      blockWhenExhausted: true
+      jmxEnabled: true
+      testOnBorrow: false
+      testOnReturn: false
+    udpPort: 4245
+    udpClientPool:
+      minIdlePerKey: 0
+      maxIdlePerKey: 40
+      maxTotalPerKey: 40
+      maxTotal: -1
+      lifo: true
+      fairness: false
+      maxWaitMillis: -1
+      minEvictableIdleTimeMillis: 1800000
+      evictorShutdownTimeoutMillis: 10000
+      softMinEvictableIdleTimeMillis: -1
+      numTestsPerEvictionRun: 3
+      timeBetweenEvictionRunsMillis: -1
+      blockWhenExhausted: true
+      jmxEnabled: true
+      testOnBorrow: false
+      testOnReturn: false
+  http:
+    host: localhost
+    ip: 0.0.0.0
+    port: 4243
+    redirect-path: /secure-me
+    strict-transport-max-age: 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: 0.0.0.0
+    port: 5244
+    timeout: 60
+    subscription-lag: 120
+    scanner-batch-size: 10000
+    flush-interval-seconds: 30
+    scanner-read-ahead: 1
+    subscription-batch-size: 5000
+    incomingBufferSize: 50000000
+    wsClientPool:
+      minIdlePerKey: 0
+      maxIdlePerKey: 40
+      maxTotalPerKey: 40
+      maxTotal: -1
+      lifo: true
+      fairness: false
+      maxWaitMillis: -1
+      minEvictableIdleTimeMillis: 1800000
+      evictorShutdownTimeoutMillis: 10000
+      softMinEvictableIdleTimeMillis: -1
+      numTestsPerEvictionRun: 3
+      timeBetweenEvictionRunsMillis: -1
+      blockWhenExhausted: true
+      jmxEnabled: true
+      testOnBorrow: false
+      testOnReturn: false

balancer/pom.xml

@@ -4,13 +4,9 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>2.0.2</version>
   </parent>
   <artifactId>timely-balancer</artifactId>
-  <description>Timely Balancer</description>
-  <properties>
-    <start-class>timely.balancer.TimelyBalancer</start-class>
-  </properties>
   <dependencies>
     <dependency>
       <groupId>com.fasterxml.jackson.core</groupId>
@@ -41,12 +37,8 @@
       <artifactId>commons-beanutils</artifactId>
     </dependency>
     <dependency>
-      <groupId>commons-logging</groupId>
-      <artifactId>commons-logging</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>io.github.classgraph</groupId>
-      <artifactId>classgraph</artifactId>
+      <groupId>io.github.lukehutch</groupId>
+      <artifactId>fast-classpath-scanner</artifactId>
     </dependency>
     <dependency>
       <groupId>io.netty</groupId>
@@ -104,18 +96,10 @@
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-minicluster</artifactId>
     </dependency>
-    <dependency>
-      <groupId>org.apache.logging.log4j</groupId>
-      <artifactId>log4j-1.2-api</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>
@@ -123,12 +107,6 @@
     <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter</artifactId>
-      <exclusions>
-        <exclusion>
-          <groupId>org.springframework.boot</groupId>
-          <artifactId>spring-boot-starter-logging</artifactId>
-        </exclusion>
-      </exclusions>
     </dependency>
     <dependency>
       <groupId>org.springframework.boot</groupId>
@@ -138,18 +116,6 @@
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-validation</artifactId>
     </dependency>
-    <dependency>
-      <groupId>org.springframework.boot</groupId>
-      <artifactId>spring-boot-starter-web</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.springframework.cloud</groupId>
-      <artifactId>spring-cloud-context</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.springframework.cloud</groupId>
-      <artifactId>spring-cloud-starter-bootstrap</artifactId>
-    </dependency>
     <dependency>
       <groupId>org.springframework.security</groupId>
       <artifactId>spring-security-config</artifactId>
@@ -190,6 +156,11 @@
       <artifactId>log4j-core</artifactId>
       <scope>runtime</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.logging.log4j</groupId>
+      <artifactId>log4j-slf4j-impl</artifactId>
+      <scope>runtime</scope>
+    </dependency>
     <dependency>
       <groupId>org.javassist</groupId>
       <artifactId>javassist</artifactId>
@@ -205,11 +176,6 @@
       <artifactId>jsoup</artifactId>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>org.springframework</groupId>
-      <artifactId>spring-test</artifactId>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-test-autoconfigure</artifactId>
@@ -218,10 +184,6 @@
   </dependencies>
   <build>
     <plugins>
-      <plugin>
-        <groupId>org.springframework.boot</groupId>
-        <artifactId>spring-boot-maven-plugin</artifactId>
-      </plugin>
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-assembly-plugin</artifactId>
@@ -281,20 +243,6 @@
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-failsafe-plugin</artifactId>
       </plugin>
-      <plugin>
-        <groupId>org.apache.maven.plugins</groupId>
-        <artifactId>maven-jar-plugin</artifactId>
-        <version>3.1.2</version>
-        <configuration>
-          <archive>
-            <manifest>
-              <!--suppress UnresolvedMavenProperty, MavenModelInspection -->
-              <mainClass>${start-class}</mainClass>
-              <addDefaultImplementationEntries>true</addDefaultImplementationEntries>
-            </manifest>
-          </archive>
-        </configuration>
-      </plugin>
       <plugin>
         <groupId>org.codehaus.mojo</groupId>
         <artifactId>build-helper-maven-plugin</artifactId>
@@ -322,17 +270,4 @@
       </extension>
     </extensions>
   </build>
-  <profiles>
-    <profile>
-      <id>docker</id>
-      <build>
-        <plugins>
-          <plugin>
-            <groupId>com.spotify</groupId>
-            <artifactId>docker-maven-plugin</artifactId>
-          </plugin>
-        </plugins>
-      </build>
-    </profile>
-  </profiles>
 </project>

balancer/src/main/dist/assembly.xml

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

balancer/src/main/docker/Dockerfile

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

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

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

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

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

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

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

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

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

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

@@ -4,15 +4,11 @@ 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;
+import timely.configuration.Http;
 
-@RefreshScope
-@ConfigurationProperties(prefix = "timely.balancer.http")
-public class BalancerHttpProperties extends HttpProperties {
+public class BalancerHttp extends Http {
 
     @Valid
     @NestedConfigurationProperty

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

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

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

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

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

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

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

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

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

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

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

@@ -7,11 +7,11 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import timely.balancer.ArrivalRate;
-import timely.balancer.configuration.BalancerProperties;
+import timely.balancer.configuration.BalancerConfiguration;
 
 public class TimelyBalancedHost {
 
-    private static final Logger log = LoggerFactory.getLogger(TimelyBalancedHost.class);
+    private static final Logger LOG = LoggerFactory.getLogger(TimelyBalancedHost.class);
 
     private String host;
     private int tcpPort;
@@ -19,7 +19,7 @@ public class TimelyBalancedHost {
     private int wsPort;
     private int udpPort;
     private boolean isUp = true;
-    private BalancerProperties balancerProperties;
+    private BalancerConfiguration balancerConfig;
     private int failures = 0;
     private int successes = 0;
     private ArrivalRate arrivalRate = null;
@@ -41,8 +41,8 @@ public class TimelyBalancedHost {
         return new TimelyBalancedHost(host, tcpPort, httpPort, wsPort, udpPort, arrivalRate);
     }
 
-    public void setBalancerProperties(BalancerProperties balancerProperties) {
-        this.balancerProperties = balancerProperties;
+    public void setBalancerConfig(BalancerConfiguration balancerConfig) {
+        this.balancerConfig = balancerConfig;
     }
 
     synchronized public boolean isUp() {
@@ -50,38 +50,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),
+            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),
+            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);
                 }
             }
         }

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

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

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

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

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

@@ -2,13 +2,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 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/UdpClientPool.java

@@ -2,13 +2,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 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

@@ -6,34 +6,27 @@ 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> {
 
     private final SSLContext sslContext;
-    private BalancerProperties balancerProperties;
-    private BalancerWebsocketProperties balancerWebsocketProperties;
-    private SslClientProperties sslClientProperties;
+    private BalancerConfiguration balancerConfig;
 
-    public WsClientFactory(BalancerProperties balancerProperties, BalancerWebsocketProperties balancerWebsocketProperties,
-                    SslClientProperties sslClientProperties, SSLContext sslContext) {
-        this.balancerProperties = balancerProperties;
-        this.balancerWebsocketProperties = balancerWebsocketProperties;
-        this.sslClientProperties = sslClientProperties;
+    public WsClientFactory(BalancerConfiguration balancerConfig, SSLContext sslContext) {
+        this.balancerConfig = balancerConfig;
         this.sslContext = sslContext;
     }
 
     @Override
     public PooledObject<WebSocketSubscriptionClient> makeObject(TimelyBalancedHost k) throws Exception {
 
-        int bufferSize = balancerWebsocketProperties.getIncomingBufferSize();
+        int bufferSize = balancerConfig.getWebsocket().getIncomingBufferSize();
         WebSocketSubscriptionClient client = new WebSocketSubscriptionClient(sslContext, k.getHost(), k.getHttpPort(), k.getWsPort(),
-                        sslClientProperties.isUseClientCert(), balancerProperties.isLoginRequired(), sslClientProperties.isHostVerificationEnabled(),
-                        bufferSize);
+                        balancerConfig.getSecurity().getClientSsl().isUseClientCert(), balancerConfig.isLoginRequired(), "", "",
+                        balancerConfig.getSecurity().getClientSsl().isHostVerificationEnabled(), bufferSize);
         return new DefaultPooledObject<>(client);
     }
 

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

@@ -4,17 +4,13 @@ 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 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

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

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

@@ -36,21 +36,21 @@ import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.http.HttpVersion;
 import io.netty.util.ReferenceCountUtil;
 import timely.api.request.AuthenticatedRequest;
-import timely.api.request.HttpRequest;
 import timely.api.request.MetricRequest;
+import timely.api.request.timeseries.HttpRequest;
 import timely.api.request.timeseries.QueryRequest;
 import timely.api.response.TimelyException;
-import timely.auth.TimelyAuthenticationToken;
 import timely.auth.util.HttpHeaderUtils;
 import timely.auth.util.ProxiedEntityUtils;
-import timely.balancer.component.MetricResolver;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.balancer.connection.http.HttpClientPool;
+import timely.balancer.resolver.MetricResolver;
 import timely.netty.http.TimelyHttpHandler;
+import timely.netty.http.auth.TimelyAuthenticationToken;
 
 public class HttpRelayHandler extends SimpleChannelInboundHandler<HttpRequest> implements TimelyHttpHandler {
 
-    private static final Logger log = LoggerFactory.getLogger(HttpRelayHandler.class);
+    private static final Logger LOG = LoggerFactory.getLogger(HttpRelayHandler.class);
     private final HttpClientPool httpClientPool;
     private MetricResolver metricResolver;
 
@@ -132,12 +132,12 @@ 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));
@@ -169,7 +169,7 @@ public class HttpRelayHandler extends SimpleChannelInboundHandler<HttpRequest> i
             if (client != null && k != null) {
                 httpClientPool.returnObject(k, client);
             } else {
-                log.error("NOT RETURNING CONNECTION! " + msg.getHttpRequest().uri());
+                LOG.error("NOT RETURNING CONNECTION! " + msg.getHttpRequest().uri());
             }
         }
     }

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

@@ -14,23 +14,24 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
 import timely.api.request.MetricRequest;
 import timely.api.request.TcpRequest;
-import timely.balancer.component.MetricResolver;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.balancer.connection.tcp.TcpClientPool;
+import timely.balancer.resolver.BalancedMetricResolver;
+import timely.balancer.resolver.MetricResolver;
 import timely.client.tcp.TcpClient;
 import timely.netty.Constants;
 
 public class TcpRelayHandler extends SimpleChannelInboundHandler<TcpRequest> {
 
-    private static final Logger log = LoggerFactory.getLogger(TcpRelayHandler.class);
+    private static final Logger LOG = LoggerFactory.getLogger(TcpRelayHandler.class);
     private static final String LOG_ERR_MSG = "Error storing put metric: {}";
     private static final String ERR_MSG = "Error storing put metric: ";
 
-    private MetricResolver metricResolver;
+    private BalancedMetricResolver metricResolver;
     private List<TcpClientPool> tcpClientPools;
 
     public TcpRelayHandler(MetricResolver metricResolver, List<TcpClientPool> tcpClientPools) {
-        this.metricResolver = metricResolver;
+        this.metricResolver = (BalancedMetricResolver) metricResolver;
         this.tcpClientPools = tcpClientPools;
     }
 
@@ -41,7 +42,7 @@ 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;
@@ -63,10 +64,10 @@ public class TcpRelayHandler extends SimpleChannelInboundHandler<TcpRequest> {
             }
 
         } catch (IOException e) {
-            log.error(LOG_ERR_MSG, msg, e);
+            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());
             }
         }
     }
@@ -83,7 +84,7 @@ public class TcpRelayHandler extends SimpleChannelInboundHandler<TcpRequest> {
                 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

@@ -12,15 +12,15 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
 import timely.api.request.MetricRequest;
 import timely.api.request.UdpRequest;
-import timely.balancer.component.MetricResolver;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.balancer.connection.udp.UdpClientPool;
+import timely.balancer.resolver.MetricResolver;
 import timely.client.udp.UdpClient;
 import timely.netty.Constants;
 
 public class UdpRelayHandler extends SimpleChannelInboundHandler<UdpRequest> {
 
-    private static final Logger log = LoggerFactory.getLogger(UdpRelayHandler.class);
+    private static final Logger LOG = LoggerFactory.getLogger(UdpRelayHandler.class);
     private static final String LOG_ERR_MSG = "Error storing put metric: {}";
     private static final String ERR_MSG = "Error storing put metric: ";
 
@@ -34,7 +34,7 @@ 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();
@@ -50,10 +50,10 @@ public class UdpRelayHandler extends SimpleChannelInboundHandler<UdpRequest> {
             }
 
         } catch (Exception e) {
-            log.error(LOG_ERR_MSG, msg, e);
+            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());
             }
         }
     }
@@ -69,7 +69,7 @@ public class UdpRelayHandler extends SimpleChannelInboundHandler<UdpRequest> {
             } 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

@@ -21,13 +21,13 @@ import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import io.netty.util.concurrent.ScheduledFuture;
 import timely.api.response.TimelyException;
-import timely.auth.TimelyAuthenticationToken;
 import timely.auth.util.ProxiedEntityUtils;
 import timely.client.websocket.ClientHandler;
+import timely.netty.http.auth.TimelyAuthenticationToken;
 
 public class WsClientHandler extends ClientHandler {
 
-    private static final Logger log = LoggerFactory.getLogger(WsClientHandler.class);
+    private static final Logger LOG = LoggerFactory.getLogger(WsClientHandler.class);
     private final ChannelHandlerContext ctx;
     private final TimelyAuthenticationToken token;
     private ScheduledFuture<?> ping;
@@ -36,7 +36,7 @@ public class WsClientHandler extends ClientHandler {
         this.ctx = ctx;
         this.token = token;
         this.ping = this.ctx.executor().scheduleAtFixedRate(() -> {
-            log.trace("Sending ping on channel {}", ctx.channel());
+            LOG.trace("Sending ping on channel {}", ctx.channel());
             ctx.writeAndFlush(new PingWebSocketFrame());
         }, pingRate, pingRate, TimeUnit.SECONDS);
     }
@@ -64,7 +64,7 @@ public class WsClientHandler extends ClientHandler {
     public void onOpen(Session session, EndpointConfig config) {
         session.addMessageHandler(String.class, message -> {
             ctx.writeAndFlush(new TextWebSocketFrame(message));
-            log.debug("Message received on Websocket session {}: {}", session.getId(), message);
+            LOG.debug("Message received on Websocket session {}: {}", session.getId(), message);
         });
     }
 
@@ -72,7 +72,7 @@ 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());
+            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);
         }
@@ -84,7 +84,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

@@ -11,9 +11,9 @@ 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 +33,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

@@ -21,26 +21,26 @@ 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 timely.api.request.subscription.AddSubscription;
+import timely.api.request.subscription.CloseSubscription;
+import timely.api.request.subscription.CreateSubscription;
+import timely.api.request.subscription.RemoveSubscription;
+import timely.api.request.subscription.SubscriptionRequest;
 import timely.api.response.TimelyException;
 import timely.auth.util.ProxiedEntityUtils;
-import timely.balancer.component.MetricResolver;
-import timely.balancer.configuration.BalancerWebsocketProperties;
+import timely.balancer.configuration.BalancerConfiguration;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.balancer.connection.ws.WsClientPool;
+import timely.balancer.resolver.MetricResolver;
 import timely.client.websocket.subscription.WebSocketSubscriptionClient;
 import timely.netty.http.TimelyHttpHandler;
 
 public class WsRelayHandler extends SimpleChannelInboundHandler<SubscriptionRequest> implements TimelyHttpHandler {
 
-    private static final Logger log = LoggerFactory.getLogger(WsRelayHandler.class);
+    private static final Logger LOG = LoggerFactory.getLogger(WsRelayHandler.class);
 
-    private BalancerWebsocketProperties balancerWebsocketProperties;
-    private WsClientPool wsClientPool;
+    private final BalancerConfiguration balancerConfig;
+    private final WsClientPool wsClientPool;
     private MetricResolver metricResolver;
     static private ObjectMapper mapper;
     Map<String,Map<String,WsClientHolder>> wsClients = new ConcurrentHashMap<>();
@@ -54,8 +54,8 @@ 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;
     }
@@ -89,7 +89,7 @@ public class WsRelayHandler extends SimpleChannelInboundHandler<SubscriptionRequ
                         if (metricToClientMap != null) {
                             synchronized (metricToClientMap) {
                                 // close all clients for this subscriptionId
-                                log.debug("Channel closed, closing subscriptions for subscriptionId:{}", currentSubscriptionId);
+                                LOG.debug("Channel closed, closing subscriptions for subscriptionId:{}", currentSubscriptionId);
                                 for (Map.Entry<String,WsClientHolder> entry : metricToClientMap.entrySet()) {
                                     entry.getValue().close(wsClientPool);
                                 }
@@ -108,7 +108,7 @@ 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);
+                    LOG.info("ChannelHandlerContext not found for subscriptionId:{} - createSubscription not called?", subscriptionId);
                     sendErrorResponse(ctx, HttpResponseStatus.BAD_REQUEST, new IllegalArgumentException("Must call create first"));
                 }
                 Map<String,WsClientHolder> metricToClientMap = wsClients.get(subscriptionId);
@@ -122,7 +122,7 @@ public class WsRelayHandler extends SimpleChannelInboundHandler<SubscriptionRequ
                     if (hostClientHolder == null) {
                         k = metricResolver.getHostPortKey(metric);
                         client = wsClientPool.borrowObject(k);
-                        client.open(new WsClientHandler(origContext, add.getToken(), (balancerWebsocketProperties.getTimeout() / 2)));
+                        client.open(new WsClientHandler(origContext, add.getToken(), (balancerConfig.getWebsocket().getTimeout() / 2)));
                         metricToClientMap.put(metric, new WsClientHolder(k, client));
                     } else {
                         client = hostClientHolder.getClient();
@@ -159,7 +159,7 @@ public class WsRelayHandler extends SimpleChannelInboundHandler<SubscriptionRequ
                         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:{} metric:{}", subscriptionId, metric);
                         }
                     }
                 }
@@ -180,11 +180,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 +198,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/BalancedMetricResolver.java

@@ -1,11 +1,11 @@
-package timely.balancer.component;
+package timely.balancer.resolver;
 
-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.balancer.component.Balancer.ASSIGNMENTS_LAST_UPDATED_PATH;
-import static timely.balancer.component.Balancer.ASSIGNMENTS_LOCK_PATH;
-import static timely.balancer.component.Balancer.LEADER_LATCH_PATH;
+import static timely.Server.SERVICE_DISCOVERY_PATH;
+import static timely.balancer.Balancer.ASSIGNMENTS_LAST_UPDATED_PATH;
+import static timely.balancer.Balancer.ASSIGNMENTS_LOCK_PATH;
+import static timely.balancer.Balancer.LEADER_LATCH_PATH;
+import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS;
+import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS_LOCK_PATH;
 
 import java.io.FileNotFoundException;
 import java.nio.charset.Charset;
@@ -29,9 +29,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
-
 import org.apache.commons.lang3.SerializationUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.curator.framework.CuratorFramework;
@@ -58,25 +55,25 @@ import org.apache.hadoop.fs.Path;
 import org.apache.zookeeper.CreateMode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.stereotype.Component;
 
 import com.csvreader.CsvReader;
 import com.csvreader.CsvWriter;
 
 import timely.ServerDetails;
 import timely.balancer.ArrivalRate;
-import timely.balancer.configuration.BalancerProperties;
+import timely.balancer.configuration.BalancerConfiguration;
 import timely.balancer.connection.TimelyBalancedHost;
-import timely.common.configuration.CacheProperties;
+import timely.balancer.healthcheck.HealthChecker;
 
-@Component
-@ConditionalOnMissingBean(MetricResolver.class)
-public class MetricResolver {
+public class BalancedMetricResolver implements MetricResolver {
 
-    private static final Logger log = LoggerFactory.getLogger(MetricResolver.class);
+    private static final Logger LOG = LoggerFactory.getLogger(BalancedMetricResolver.class);
+    private Map<String,TimelyBalancedHost> metricToHostMap = new TreeMap<>();
+    private Map<String,ArrivalRate> metricMap = new HashMap<>();
     private ReentrantReadWriteLock balancerLock = new ReentrantReadWriteLock();
+    private List<TimelyBalancedHost> serverList = new ArrayList<>();
     private Random r = new Random();
+    final private HealthChecker healthChecker;
     private ScheduledExecutorService assignmentExecutor = Executors.newScheduledThreadPool(2);
     private ScheduledExecutorService arrivalRateExecutor = Executors.newScheduledThreadPool(2);
     private int roundRobinCounter = 0;
@@ -84,44 +81,36 @@ public class MetricResolver {
     private DistributedAtomicValue nonCachedMetricsIP;
     private ReentrantReadWriteLock nonCachedMetricsLocalLock = new ReentrantReadWriteLock();
     private InterProcessReadWriteLock nonCachedMetricsIPRWLock;
+    private BalancerConfiguration balancerConfig;
     private LeaderLatch leaderLatch;
     private AtomicBoolean isLeader = new AtomicBoolean(false);
     private InterProcessReadWriteLock assignmentsIPRWLock;
     private DistributedAtomicLong assignmentsLastUpdatedInHdfs;
     private AtomicLong assignmentsLastUpdatedLocal = new AtomicLong(0);
-    protected List<TimelyBalancedHost> serverList = new ArrayList<>();
-    protected Map<String,TimelyBalancedHost> metricToHostMap = new TreeMap<>();
-    protected Map<String,ArrivalRate> metricMap = new HashMap<>();
-    final protected CuratorFramework curatorFramework;
-    final protected BalancerProperties balancerProperties;
-    final protected CacheProperties cacheProperties;
-    final protected HealthChecker healthChecker;
 
     private enum BalanceType {
         HIGH_LOW, HIGH_AVG, AVG_LOW
     }
 
-    public MetricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
-                    HealthChecker healthChecker) {
-        this.curatorFramework = curatorFramework;
-        this.balancerProperties = balancerProperties;
-        this.cacheProperties = cacheProperties;
+    public BalancedMetricResolver(CuratorFramework curatorFramework, BalancerConfiguration balancerConfig, HealthChecker healthChecker) {
+        this.balancerConfig = balancerConfig;
         this.healthChecker = healthChecker;
-    }
 
-    @PostConstruct
-    public void setup() {
         assignmentsIPRWLock = new InterProcessReadWriteLock(curatorFramework, ASSIGNMENTS_LOCK_PATH);
         testIPRWLock(curatorFramework, assignmentsIPRWLock, ASSIGNMENTS_LOCK_PATH);
         startLeaderLatch(curatorFramework);
         startServiceListener(curatorFramework);
         assignmentsLastUpdatedInHdfs = new DistributedAtomicLong(curatorFramework, ASSIGNMENTS_LAST_UPDATED_PATH, new RetryForever(1000));
 
-        TreeCacheListener assignmentListener = (curframework, event) -> {
-            log.debug("Handling assignments event {}. assignmentsLastUpdatedInHdfs:{}", event.getType().toString(),
-                            new Date(assignmentsLastUpdatedInHdfs.get().postValue()));
-            if (event.getType().equals(TreeCacheEvent.Type.NODE_UPDATED)) {
-                readAssignmentsFromHdfs(true);
+        TreeCacheListener assignmentListener = new TreeCacheListener() {
+
+            @Override
+            public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent event) throws Exception {
+                LOG.debug("Handling assignments event {}. assignmentsLastUpdatedInHdfs:{}", event.getType().toString(),
+                                new Date(assignmentsLastUpdatedInHdfs.get().postValue()));
+                if (event.getType().equals(TreeCacheEvent.Type.NODE_UPDATED)) {
+                    readAssignmentsFromHdfs(true);
+                }
             }
         };
 
@@ -129,16 +118,20 @@ public class MetricResolver {
             assignmentTreeCache.getListenable().addListener(assignmentListener);
             assignmentTreeCache.start();
         } catch (Exception e) {
-            log.error(e.getMessage(), e);
+            LOG.error(e.getMessage(), e);
         }
 
         nonCachedMetricsIPRWLock = new InterProcessReadWriteLock(curatorFramework, NON_CACHED_METRICS_LOCK_PATH);
         testIPRWLock(curatorFramework, nonCachedMetricsIPRWLock, NON_CACHED_METRICS_LOCK_PATH);
         nonCachedMetricsIP = new DistributedAtomicValue(curatorFramework, NON_CACHED_METRICS, new RetryForever(1000));
-        TreeCacheListener nonCachedMetricsListener = (framework, event) -> {
-            if (event.getType().equals(TreeCacheEvent.Type.NODE_UPDATED)) {
-                log.debug("Handling nonCachedMetricsIP event {}", event.getType().toString());
-                readNonCachedMetricsIP();
+        TreeCacheListener nonCachedMetricsListener = new TreeCacheListener() {
+
+            @Override
+            public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent event) throws Exception {
+                if (event.getType().equals(TreeCacheEvent.Type.NODE_UPDATED)) {
+                    LOG.debug("Handling nonCachedMetricsIP event {}", event.getType().toString());
+                    readNonCachedMetricsIP();
+                }
             }
         };
 
@@ -146,12 +139,12 @@ public class MetricResolver {
             nonCachedMetricsTreeCache.getListenable().addListener(nonCachedMetricsListener);
             nonCachedMetricsTreeCache.start();
         } catch (Exception e) {
-            log.error(e.getMessage(), e);
+            LOG.error(e.getMessage(), e);
         }
 
         nonCachedMetricsLocalLock.writeLock().lock();
         try {
-            addNonCachedMetrics(cacheProperties.getNonCachedMetrics());
+            addNonCachedMetrics(balancerConfig.getCache().getNonCachedMetrics());
         } finally {
             nonCachedMetricsLocalLock.writeLock().unlock();
         }
@@ -164,7 +157,7 @@ public class MetricResolver {
                 try {
                     balance();
                 } catch (Exception e) {
-                    log.error(e.getMessage(), e);
+                    LOG.error(e.getMessage(), e);
                 }
             }
         }, 15, 15, TimeUnit.MINUTES);
@@ -175,48 +168,22 @@ public class MetricResolver {
                     long lastLocalUpdate = assignmentsLastUpdatedLocal.get();
                     long lastHdfsUpdate = assignmentsLastUpdatedInHdfs.get().postValue();
                     if (lastLocalUpdate > lastHdfsUpdate) {
-                        log.debug("Leader writing assignments to hdfs lastLocalUpdate ({}) > lastHdfsUpdate ({})", new Date(lastLocalUpdate),
+                        LOG.debug("Leader writing assignments to hdfs lastLocalUpdate ({}) > lastHdfsUpdate ({})", new Date(lastLocalUpdate),
                                         new Date(lastHdfsUpdate));
                         writeAssignmentsToHdfs();
                     } else {
-                        log.trace("Leader not writing assignments to hdfs lastLocalUpdate ({}) <= lastHdfsUpdate ({})", new Date(lastLocalUpdate),
+                        LOG.trace("Leader not writing assignments to hdfs lastLocalUpdate ({}) <= lastHdfsUpdate ({})", new Date(lastLocalUpdate),
                                         new Date(lastHdfsUpdate));
                     }
                 } else {
                     readAssignmentsFromHdfs(true);
                 }
             } catch (Exception e) {
-                log.error(e.getMessage(), e);
+                LOG.error(e.getMessage(), e);
             }
         }, 10, 60, TimeUnit.SECONDS);
     }
 
-    @PreDestroy
-    public void close() throws Exception {
-
-        this.assignmentExecutor.shutdown();
-        try {
-            this.assignmentExecutor.awaitTermination(5, TimeUnit.SECONDS);
-        } catch (InterruptedException e) {
-
-        } finally {
-            if (!this.assignmentExecutor.isTerminated()) {
-                this.assignmentExecutor.shutdownNow();
-            }
-        }
-
-        this.arrivalRateExecutor.shutdown();
-        try {
-            this.arrivalRateExecutor.awaitTermination(5, TimeUnit.SECONDS);
-        } catch (InterruptedException e) {
-
-        } finally {
-            if (!this.arrivalRateExecutor.isTerminated()) {
-                this.arrivalRateExecutor.shutdownNow();
-            }
-        }
-    }
-
     private void readNonCachedMetricsIP() {
 
         Set<String> currentNonCachedMetricsDistributed = new TreeSet<>();
@@ -230,31 +197,31 @@ public class MetricResolver {
                 try {
                     currentNonCachedMetricsDistributed = SerializationUtils.deserialize(currentNonCachedMetricsDistributedBytes);
                 } catch (Exception e) {
-                    log.error(e.getMessage(), e);
+                    LOG.error(e.getMessage(), e);
                     currentNonCachedMetricsDistributed = new TreeSet<>();
                 }
             }
         } catch (Exception e) {
-            log.error(e.getMessage(), e);
+            LOG.error(e.getMessage(), e);
         } finally {
             try {
                 nonCachedMetricsIPRWLock.readLock().release();
             } catch (Exception e) {
-                log.error(e.getMessage(), e);
+                LOG.error(e.getMessage(), e);
             }
         }
 
         nonCachedMetricsLocalLock.writeLock().lock();
         try {
             if (nonCachedMetrics.containsAll(currentNonCachedMetricsDistributed)) {
-                log.debug("local nonCachedMetrics already contains {}", currentNonCachedMetricsDistributed);
+                LOG.debug("local nonCachedMetrics already contains {}", currentNonCachedMetricsDistributed);
             } else {
                 currentNonCachedMetricsDistributed.removeAll(nonCachedMetrics);
-                log.debug("Adding {} to local nonCachedMetrics", currentNonCachedMetricsDistributed);
+                LOG.debug("Adding {} to local nonCachedMetrics", currentNonCachedMetricsDistributed);
                 nonCachedMetrics.addAll(currentNonCachedMetricsDistributed);
             }
         } catch (Exception e) {
-            log.error(e.getMessage(), e);
+            LOG.error(e.getMessage(), e);
         } finally {
             nonCachedMetricsLocalLock.writeLock().unlock();
         }
@@ -269,19 +236,19 @@ public class MetricResolver {
 
                 @Override
                 public void isLeader() {
-                    log.info("this balancer is the leader");
+                    LOG.info("this balancer is the leader");
                     isLeader.set(true);
                     writeAssignmentsToHdfs();
                 }
 
                 @Override
                 public void notLeader() {
-                    log.info("this balancer is not the leader");
+                    LOG.info("this balancer is not the leader");
                     isLeader.set(false);
                 }
             });
         } catch (Exception e) {
-            log.error(e.getMessage(), e);
+            LOG.error(e.getMessage(), e);
         }
     }
 
@@ -298,9 +265,9 @@ public class MetricResolver {
                     ServerDetails pl = si.getPayload();
                     TimelyBalancedHost tbh = TimelyBalancedHost.of(pl.getHost(), pl.getTcpPort(), pl.getHttpPort(), pl.getWsPort(), pl.getUdpPort(),
                                     new ArrivalRate(arrivalRateExecutor));
-                    log.info("adding service {} host:{} tcpPort:{} httpPort:{} wsPort:{} udpPort:{}", si.getId(), pl.getHost(), pl.getTcpPort(),
+                    LOG.info("adding service {} host:{} tcpPort:{} httpPort:{} wsPort:{} udpPort:{}", si.getId(), pl.getHost(), pl.getTcpPort(),
                                     pl.getHttpPort(), pl.getWsPort(), pl.getUdpPort());
-                    tbh.setBalancerProperties(balancerProperties);
+                    tbh.setBalancerConfig(balancerConfig);
                     serverList.add(tbh);
                 }
                 healthChecker.setTimelyHosts(serverList);
@@ -319,10 +286,10 @@ public class MetricResolver {
                         List<ServiceInstance<ServerDetails>> instances = serviceCache.getInstances();
                         Set<TimelyBalancedHost> availableHosts = new HashSet<>();
                         for (ServiceInstance<ServerDetails> si : instances) {
-                            ServerDetails pl = si.getPayload();
+                            ServerDetails pl = (ServerDetails) si.getPayload();
                             TimelyBalancedHost tbh = TimelyBalancedHost.of(pl.getHost(), pl.getTcpPort(), pl.getHttpPort(), pl.getWsPort(), pl.getUdpPort(),
                                             new ArrivalRate(arrivalRateExecutor));
-                            tbh.setBalancerProperties(balancerProperties);
+                            tbh.setBalancerConfig(balancerConfig);
                             availableHosts.add(tbh);
                         }
 
@@ -335,7 +302,7 @@ public class MetricResolver {
                                 availableHosts.remove(h);
                             } else {
                                 itr.remove();
-                                log.info("removing service {}:{} host:{} tcpPort:{} httpPort:{} wsPort:{} udpPort:{}", h.getHost(), h.getTcpPort(), h.getHost(),
+                                LOG.info("removing service {}:{} host:{} tcpPort:{} httpPort:{} wsPort:{} udpPort:{}", h.getHost(), h.getTcpPort(), h.getHost(),
                                                 h.getTcpPort(), h.getHttpPort(), h.getWsPort(), h.getUdpPort());
                                 for (Map.Entry<String,TimelyBalancedHost> e : metricToHostMap.entrySet()) {
                                     if (e.getValue().equals(h)) {
@@ -348,7 +315,7 @@ public class MetricResolver {
 
                         // add new hosts that were not previously known
                         for (TimelyBalancedHost h : availableHosts) {
-                            log.info("adding service {}:{} host:{} tcpPort:{} httpPort:{} wsPort:{} udpPort:{}", h.getHost(), h.getTcpPort(), h.getHost(),
+                            LOG.info("adding service {}:{} host:{} tcpPort:{} httpPort:{} wsPort:{} udpPort:{}", h.getHost(), h.getTcpPort(), h.getHost(),
                                             h.getTcpPort(), h.getHttpPort(), h.getWsPort(), h.getUdpPort());
                             serverList.add(h);
                             rebalanceNeeded = true;
@@ -359,7 +326,7 @@ public class MetricResolver {
                             for (String s : reassignMetrics) {
                                 TimelyBalancedHost h = getRoundRobinHost(null);
                                 assignMetric(s, h);
-                                log.debug("Assigned server removed.  Assigning {} to server {}:{}", s, h.getTcpPort());
+                                LOG.debug("Assigned server removed.  Assigning {} to server {}:{}", s, h.getTcpPort());
                             }
                         }
 
@@ -376,14 +343,14 @@ public class MetricResolver {
 
                 @Override
                 public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
-                    log.info("serviceCache state changed.  Connected:{}", connectionState.isConnected());
+                    LOG.info("serviceCache state changed.  Connected:{}", connectionState.isConnected());
                 }
             };
             serviceCache.addListener(listener);
             serviceCache.start();
 
         } catch (Exception e) {
-            log.error(e.getMessage(), e);
+            LOG.error(e.getMessage(), e);
         }
     }
 
@@ -495,7 +462,7 @@ public class MetricResolver {
 
         int numReassigned = 0;
         if (isLeader.get()) {
-            double controlBandPercentage = balancerProperties.getControlBandPercentage();
+            double controlBandPercentage = balancerConfig.getControlBandPercentage();
             // save current rates so that we can modify
             double totalArrivalRate = 0;
             Map<TimelyBalancedHost,Double> calculatedRates = new HashMap<>();
@@ -527,19 +494,19 @@ public class MetricResolver {
             }
 
             if (highHosts.isEmpty() && lowHosts.isEmpty()) {
-                log.debug("Host's arrival rates are within {} of the average", controlBandPercentage);
+                LOG.debug("Host's arrival rates are within {} of the average", controlBandPercentage);
             } else if (!highHosts.isEmpty() && !lowHosts.isEmpty()) {
-                log.debug("begin rebalancing {}", BalanceType.HIGH_LOW);
+                LOG.debug("begin rebalancing {}", BalanceType.HIGH_LOW);
                 numReassigned = rebalance(highHosts, lowHosts, calculatedRates, averageArrivalRate, BalanceType.HIGH_LOW);
-                log.debug("end rebalancing {} - reassigned {}", BalanceType.HIGH_LOW, numReassigned);
+                LOG.debug("end rebalancing {} - reassigned {}", BalanceType.HIGH_LOW, numReassigned);
             } else if (lowHosts.isEmpty()) {
-                log.debug("begin rebalancing {}", BalanceType.HIGH_AVG);
+                LOG.debug("begin rebalancing {}", BalanceType.HIGH_AVG);
                 numReassigned = rebalance(highHosts, avgHosts, calculatedRates, averageArrivalRate, BalanceType.HIGH_AVG);
-                log.debug("end rebalancing {} - reassigned {}", BalanceType.HIGH_AVG, numReassigned);
+                LOG.debug("end rebalancing {} - reassigned {}", BalanceType.HIGH_AVG, numReassigned);
             } else {
-                log.debug("begin rebalancing {}", BalanceType.AVG_LOW);
+                LOG.debug("begin rebalancing {}", BalanceType.AVG_LOW);
                 numReassigned = rebalance(avgHosts, lowHosts, calculatedRates, averageArrivalRate, BalanceType.AVG_LOW);
-                log.debug("end rebalancing {} - reassigned {}", BalanceType.AVG_LOW, numReassigned);
+                LOG.debug("end rebalancing {} - reassigned {}", BalanceType.AVG_LOW, numReassigned);
             }
         }
         return numReassigned;
@@ -582,7 +549,7 @@ public class MetricResolver {
                         }
                     }
 
-                    log.trace("focusHost {}:{} desiredChange:{} rateSortedMetrics.size():{}", h.getHost(), h.getTcpPort(), desiredChange,
+                    LOG.trace("focusHost {}:{} desiredChange:{} rateSortedMetrics.size():{}", h.getHost(), h.getTcpPort(), desiredChange,
                                     rateSortedMetrics.size());
 
                     boolean doneWithHost = false;
@@ -594,19 +561,19 @@ public class MetricResolver {
                             while (e == null && itr.hasNext()) {
                                 e = itr.next();
                                 if (e.getKey() > desiredChange) {
-                                    log.trace("Skipping:{} rate:{}", e.getValue(), e.getKey());
+                                    LOG.trace("Skipping:{} rate:{}", e.getValue(), e.getKey());
                                     e = null;
                                 }
                             }
 
                             if (e == null) {
-                                log.trace("no metric small enough");
+                                LOG.trace("no metric small enough");
                                 doneWithHost = true;
                             } else {
-                                log.trace("Selected:{} rate:{}", e.getValue(), e.getKey());
+                                LOG.trace("Selected:{} rate:{}", e.getValue(), e.getKey());
                                 TimelyBalancedHost candidateHost = chooseHost(selectFromHosts, calculatedRates, calculatedRates.get(h), balanceType);
                                 if (candidateHost == null) {
-                                    log.trace("candidate host is null");
+                                    LOG.trace("candidate host is null");
                                     doneWithHost = true;
                                 } else {
                                     String metric = e.getValue();
@@ -619,24 +586,24 @@ public class MetricResolver {
                                     // don't move this metric again this host or balance
                                     itr.remove();
                                     tempMetricMap.remove(metric);
-                                    log.info("rebalancing: reassigning metric:{} rate:{} from server {}:{} to {}:{} remaining delta {}", metric, metricRate,
+                                    LOG.info("rebalancing: reassigning metric:{} rate:{} from server {}:{} to {}:{} remaining delta {}", metric, metricRate,
                                                     h.getHost(), h.getTcpPort(), candidateHost.getHost(), candidateHost.getTcpPort(), desiredChange);
                                 }
                             }
                         }
                         if (!itr.hasNext()) {
-                            log.trace("Reached end or rateSortedMetrics");
+                            LOG.trace("Reached end or rateSortedMetrics");
                             doneWithHost = true;
                         }
                         if (balanceType.equals(BalanceType.HIGH_LOW) || balanceType.equals(BalanceType.HIGH_AVG)) {
                             if (calculatedRates.get(h) <= targetArrivalRate) {
                                 doneWithHost = true;
-                                log.trace("calculatedRates.get(h) <= targetArivalRate");
+                                LOG.trace("calculatedRates.get(h) <= targetArivalRate");
                             }
                         } else {
                             if (calculatedRates.get(h) >= targetArrivalRate) {
                                 doneWithHost = true;
-                                log.trace("calculatedRates.get(h) >= targetArivalRate");
+                                LOG.trace("calculatedRates.get(h) >= targetArivalRate");
                             }
                         }
                     }
@@ -651,7 +618,7 @@ public class MetricResolver {
     private void addNonCachedMetrics(Collection<String> nonCachedMetricsUpdate) {
         if (!nonCachedMetricsUpdate.isEmpty()) {
             try {
-                log.debug("Adding {} to local nonCachedMetrics", nonCachedMetricsUpdate);
+                LOG.debug("Adding {} to local nonCachedMetrics", nonCachedMetricsUpdate);
                 nonCachedMetricsLocalLock.writeLock().lock();
                 try {
                     nonCachedMetrics.addAll(nonCachedMetricsUpdate);
@@ -672,10 +639,10 @@ public class MetricResolver {
                         currentNonCachedMetricsIP = SerializationUtils.deserialize(currentNonCachedMetricsDistributedBytes);
                     }
                     if (currentNonCachedMetricsIP.containsAll(nonCachedMetricsUpdate)) {
-                        log.debug("nonCachedMetricsIP already contains {}", nonCachedMetricsUpdate);
+                        LOG.debug("nonCachedMetricsIP already contains {}", nonCachedMetricsUpdate);
                     } else {
                         nonCachedMetricsUpdate.removeAll(currentNonCachedMetricsIP);
-                        log.debug("Adding {} to nonCachedMetricsIP", nonCachedMetricsUpdate);
+                        LOG.debug("Adding {} to nonCachedMetricsIP", nonCachedMetricsUpdate);
                         TreeSet<String> updateSet = new TreeSet<>();
                         updateSet.addAll(currentNonCachedMetricsIP);
                         updateSet.addAll(nonCachedMetricsUpdate);
@@ -689,7 +656,7 @@ public class MetricResolver {
                     nonCachedMetricsIPRWLock.writeLock().release();
                 }
             } catch (Exception e) {
-                log.error(e.getMessage(), e);
+                LOG.error(e.getMessage(), e);
             }
         }
     }
@@ -702,13 +669,13 @@ public class MetricResolver {
                 curatorFramework.delete().deletingChildrenIfNeeded().forPath(path);
                 curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path);
             } catch (Exception e2) {
-                log.info(e2.getMessage(), e2);
+                LOG.info(e2.getMessage(), e2);
             }
         } finally {
             try {
                 lock.writeLock().release();
             } catch (Exception e3) {
-                log.error(e3.getMessage(), e3);
+                LOG.error(e3.getMessage(), e3);
             }
         }
     }
@@ -739,7 +706,7 @@ public class MetricResolver {
             try {
                 for (String r : nonCachedMetrics) {
                     if (metricName.matches(r)) {
-                        log.debug("Adding {} to list of non-cached metrics", metricName);
+                        LOG.debug("Adding {} to list of non-cached metrics", metricName);
                         addNonCachedMetrics(Collections.singleton(metricName));
                         return false;
                     }
@@ -751,6 +718,7 @@ public class MetricResolver {
         }
     }
 
+    @Override
     public TimelyBalancedHost getHostPortKeyIngest(String metric) {
 
         TimelyBalancedHost tbh = null;
@@ -795,7 +763,7 @@ public class MetricResolver {
                 } else if (!tbh.isUp()) {
                     TimelyBalancedHost oldTbh = tbh;
                     tbh = getLeastUsedHost();
-                    log.debug("rebalancing from host that is down: reassigning metric {} from server {}:{} to {}:{}", metric, oldTbh.getHost(),
+                    LOG.debug("rebalancing from host that is down: reassigning metric {} from server {}:{} to {}:{}", metric, oldTbh.getHost(),
                                     oldTbh.getTcpPort(), tbh.getHost(), tbh.getTcpPort());
                     if (!balancerLock.isWriteLockedByCurrentThread()) {
                         balancerLock.readLock().unlock();
@@ -834,6 +802,7 @@ public class MetricResolver {
         return tbh;
     }
 
+    @Override
     public TimelyBalancedHost getHostPortKey(String metric) {
         TimelyBalancedHost tbh = null;
 
@@ -899,15 +868,15 @@ public class MetricResolver {
         return tbh;
     }
 
-    protected void createAssignmentFile() {
+    private void createAssignmentFile() {
         try {
-            log.info("Creating assignment file: " + balancerProperties.getAssignmentFile());
+            LOG.info("Creating assignment file: " + balancerConfig.getAssignmentFile());
             Configuration configuration = new Configuration();
             FileSystem fs = FileSystem.get(configuration);
-            Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
+            Path assignmentFile = new Path(balancerConfig.getAssignmentFile());
             fs.create(assignmentFile);
         } catch (Exception e) {
-            log.error(e.getMessage(), e);
+            LOG.error(e.getMessage(), e);
         }
     }
 
@@ -918,14 +887,14 @@ public class MetricResolver {
             long lastHdfsUpdate = assignmentsLastUpdatedInHdfs.get().postValue();
             if (checkIfNecessary) {
                 if (lastHdfsUpdate <= lastLocalUpdate) {
-                    log.debug("Not reading assignments from hdfs lastHdfsUpdate ({}) <= lastLocalUpdate ({})", new Date(lastHdfsUpdate),
+                    LOG.debug("Not reading assignments from hdfs lastHdfsUpdate ({}) <= lastLocalUpdate ({})", new Date(lastHdfsUpdate),
                                     new Date(lastLocalUpdate));
                     return;
                 }
             }
-            log.debug("Reading assignments from hdfs lastHdfsUpdate ({}) > lastLocalUpdate ({})", new Date(lastHdfsUpdate), new Date(lastLocalUpdate));
+            LOG.debug("Reading assignments from hdfs lastHdfsUpdate ({}) > lastLocalUpdate ({})", new Date(lastHdfsUpdate), new Date(lastLocalUpdate));
         } catch (Exception e) {
-            log.error(e.getMessage(), e);
+            LOG.error(e.getMessage(), e);
         }
 
         // proceed with reading from HDFS
@@ -939,8 +908,9 @@ public class MetricResolver {
             balancerLock.writeLock().lock();
             Configuration configuration = new Configuration();
             FileSystem fs = FileSystem.get(configuration);
-            Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
+            Path assignmentFile = new Path(balancerConfig.getAssignmentFile());
             FSDataInputStream iStream = fs.open(assignmentFile);
+
             CsvReader reader = new CsvReader(iStream, ',', Charset.forName("UTF-8"));
             reader.setUseTextQualifier(false);
 
@@ -959,7 +929,7 @@ public class MetricResolver {
                     if (tbh == null) {
                         tbh = getRoundRobinHost(null);
                     } else {
-                        log.trace("Found assigment: {} to {}:{}", metric, host, tcpPort);
+                        LOG.trace("Found assigment: {} to {}:{}", metric, host, tcpPort);
                     }
                     assignedMetricToHostMap.put(metric, tbh);
                 }
@@ -973,7 +943,7 @@ public class MetricResolver {
                     }
                 } else {
                     Exception e1 = new IllegalStateException("Bad assignment metric:" + e.getKey() + " host:" + e.getValue());
-                    log.warn(e1.getMessage(), e1);
+                    LOG.warn(e1.getMessage(), e1);
                 }
             }
 
@@ -993,19 +963,19 @@ public class MetricResolver {
             metricMap.entrySet().removeIf(e -> !metricToHostMap.containsKey(e.getKey()));
 
             assignmentsLastUpdatedLocal.set(assignmentsLastUpdatedInHdfs.get().postValue());
-            log.info("Read {} assignments from hdfs lastHdfsUpdate = lastLocalUpdate ({})", metricToHostMap.size(),
+            LOG.info("Read {} assignments from hdfs lastHdfsUpdate = lastLocalUpdate ({})", metricToHostMap.size(),
                             new Date(assignmentsLastUpdatedLocal.get()));
         } catch (FileNotFoundException e) {
             createAssignmentFile();
             metricToHostMap.clear();
         } catch (Exception e) {
-            log.error(e.getMessage(), e);
+            LOG.error(e.getMessage(), e);
         } finally {
             balancerLock.writeLock().unlock();
             try {
                 assignmentsIPRWLock.readLock().release();
             } catch (Exception e) {
-                log.error(e.getMessage(), e);
+                LOG.error(e.getMessage(), e);
             }
         }
     }
@@ -1024,7 +994,7 @@ public class MetricResolver {
                 if (!metricToHostMap.isEmpty()) {
                     Configuration configuration = new Configuration();
                     FileSystem fs = FileSystem.get(configuration);
-                    Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
+                    Path assignmentFile = new Path(balancerConfig.getAssignmentFile());
                     if (!fs.exists(assignmentFile.getParent())) {
                         fs.mkdirs(assignmentFile.getParent());
                     }
@@ -1041,7 +1011,7 @@ public class MetricResolver {
                             writer.write(e.getValue().getHost());
                             writer.write(Integer.toString(e.getValue().getTcpPort()));
                             writer.endRecord();
-                            log.trace("Saving assigment: {} to {}:{}", e.getKey(), e.getValue().getHost(), e.getValue().getTcpPort());
+                            LOG.trace("Saving assigment: {} to {}:{}", e.getKey(), e.getValue().getHost(), e.getValue().getTcpPort());
                         }
                     }
 
@@ -1051,7 +1021,7 @@ public class MetricResolver {
                     if (!assignmentsLastUpdatedInHdfs.get().succeeded()) {
                         assignmentsLastUpdatedInHdfs.forceSet(now);
                     }
-                    log.debug("Wrote {} assignments to hdfs lastHdfsUpdate = lastLocalUpdate ({})", metricToHostMap.size(),
+                    LOG.debug("Wrote {} assignments to hdfs lastHdfsUpdate = lastLocalUpdate ({})", metricToHostMap.size(),
                                     new Date(assignmentsLastUpdatedLocal.get()));
                     // remove metric from metricMap (ArrivalRate) if not being cached
                     metricMap.entrySet().removeIf(e -> !metricToHostMap.containsKey(e.getKey()));
@@ -1061,7 +1031,7 @@ public class MetricResolver {
                 balancerLock.readLock().unlock();
             }
         } catch (Exception e) {
-            log.error(e.getMessage(), e);
+            LOG.error(e.getMessage(), e);
         } finally {
             if (writer != null) {
                 writer.close();
@@ -1069,12 +1039,12 @@ public class MetricResolver {
             try {
                 assignmentsIPRWLock.writeLock().release();
             } catch (Exception e) {
-                log.error(e.getMessage(), e);
+                LOG.error(e.getMessage(), e);
             }
         }
     }
 
-    protected void assignMetric(String metric, TimelyBalancedHost tbh) {
+    private void assignMetric(String metric, TimelyBalancedHost tbh) {
         if (isLeader.get()) {
             if (StringUtils.isNotBlank(metric) && tbh != null) {
                 balancerLock.writeLock().lock();
@@ -1086,8 +1056,40 @@ public class MetricResolver {
                 }
             } else {
                 Exception e = new IllegalStateException("Bad assignment metric:" + metric + " host:" + tbh);
-                log.warn(e.getMessage(), e);
+                LOG.warn(e.getMessage(), e);
             }
         }
     }
+
+    @Override
+    public void close() throws Exception {
+
+        this.assignmentExecutor.shutdown();
+        try {
+            this.assignmentExecutor.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+
+        } finally {
+            if (!this.assignmentExecutor.isTerminated()) {
+                this.assignmentExecutor.shutdownNow();
+            }
+        }
+
+        this.arrivalRateExecutor.shutdown();
+        try {
+            this.arrivalRateExecutor.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+
+        } finally {
+            if (!this.arrivalRateExecutor.isTerminated()) {
+                this.arrivalRateExecutor.shutdownNow();
+            }
+        }
+
+        try {
+            this.healthChecker.close();
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+        }
+    }
 }

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

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

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

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

balancer/src/main/spotbugs/excludes.xml

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <FindBugsFilter>
   <Match>
-    <Class name="timely.balancer.component.Balancer" />
+    <Class name="timely.balancer.Balancer" />
     <Method name="shutdown" />
     <Bug pattern="UC_USELESS_OBJECT" />
   </Match>

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

balancer/src/test/java/timely/balancer/test/BalancerTestRule.java

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

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

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

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

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

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

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

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

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

balancer/src/test/resources/log4j2.yml

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

client/pom.xml

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

client/src/main/java/timely/api/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/client/tcp/TcpClient.java

@@ -4,14 +4,13 @@ import java.io.BufferedWriter;
 import java.io.IOException;
 import java.io.OutputStreamWriter;
 import java.net.Socket;
-import java.nio.charset.StandardCharsets;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class TcpClient implements AutoCloseable {
 
-    private static final Logger log = LoggerFactory.getLogger(TcpClient.class);
+    private static final Logger LOG = LoggerFactory.getLogger(TcpClient.class);
 
     private final String host;
     private final int port;
@@ -76,7 +75,7 @@ public class TcpClient implements AutoCloseable {
      */
     @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;
             }
@@ -101,17 +100,17 @@ public class TcpClient implements AutoCloseable {
                 try {
                     connectTime = System.currentTimeMillis();
                     sock = new Socket(host, port);
-                    osw = new OutputStreamWriter(sock.getOutputStream(), StandardCharsets.UTF_8);
+                    osw = new OutputStreamWriter(sock.getOutputStream());
                     out = new BufferedWriter(osw);
                     backoff = 2000;
-                    log.trace("Connected to Timely at {}:{}", host, port);
+                    LOG.trace("Connected to Timely at {}:{}", host, port);
                 } catch (Exception e) {
-                    log.error("Error connecting to Timely at {}:{} - {}", host, port, e.getMessage());
+                    LOG.error("Error connecting to Timely at {}:{} - {}", host, port, e.getMessage());
                     if (sock != null) {
                         try {
                             sock.close();
                         } catch (IOException e1) {
-                            log.error(e1.getMessage());
+                            LOG.error(e1.getMessage());
                         } finally {
                             sock = null;
                         }
@@ -120,24 +119,24 @@ public class TcpClient implements AutoCloseable {
                         try {
                             osw.close();
                         } catch (IOException e1) {
-                            log.error(e1.getMessage());
+                            LOG.error(e1.getMessage());
                         }
                     }
                     if (out != null) {
                         try {
                             out.close();
                         } catch (IOException e1) {
-                            log.error(e1.getMessage());
+                            LOG.error(e1.getMessage());
                         } finally {
                             out = null;
                         }
                     }
                     backoff = backoff * 2;
-                    log.info("Will retry connection in {} ms.", backoff);
+                    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,7 +38,7 @@ 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);
     }
 

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

@@ -13,24 +13,24 @@ 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());
+        LOG.info("Websocket session {} opened.", session.getId());
         session.addMessageHandler(String.class, message -> {
-            log.info("Message received on Websocket session {}: {}", session.getId(), 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/WebSocketClient.java

@@ -10,13 +10,16 @@ import javax.net.ssl.SSLContext;
 import javax.websocket.DeploymentException;
 import javax.websocket.Session;
 
+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;
@@ -27,17 +30,21 @@ 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;
@@ -47,26 +54,35 @@ public class WebSocketClient implements AutoCloseable {
     protected volatile boolean closed = true;
 
     protected WebSocketClient(SSLContext ssl, String timelyHostname, int timelyHttpsPort, int timelyWssPort, boolean clientAuth, boolean doLogin,
-                    boolean hostVerificationEnabled, int bufferSize) {
+                    String timelyUsername, String timelyPassword, boolean hostVerificationEnabled, int bufferSize) {
         this.ssl = ssl;
         this.timelyHostname = timelyHostname;
         this.timelyHttpsPort = timelyHttpsPort;
         this.timelyWssPort = timelyWssPort;
         this.doLogin = doLogin;
         this.clientAuth = clientAuth;
+        this.timelyUsername = timelyUsername;
+        this.timelyPassword = timelyPassword;
         this.hostVerificationEnabled = hostVerificationEnabled;
         this.bufferSize = bufferSize;
 
         Preconditions.checkNotNull(timelyHostname, "%s must be supplied", "Timely host name");
         Preconditions.checkNotNull(timelyHttpsPort, "%s must be supplied", "Timely HTTPS port");
         Preconditions.checkNotNull(timelyWssPort, "%s must be supplied", "Timely WSS port");
+
+        if (doLogin && !clientAuth) {
+            if ((StringUtils.isEmpty(timelyUsername) && !StringUtils.isEmpty(timelyPassword)
+                            || (!StringUtils.isEmpty(timelyUsername) && StringUtils.isEmpty(timelyPassword)))) {
+                throw new IllegalArgumentException("Both Timely username and password must be empty or non-empty");
+            }
+        }
     }
 
-    protected WebSocketClient(String timelyHostname, int timelyHttpsPort, int timelyWssPort, boolean clientAuth, boolean doLogin, String keyStoreFile,
-                    String keyStoreType, String keyStorePass, String trustStoreFile, String trustStoreType, String trustStorePass,
-                    boolean hostVerificationEnabled, int bufferSize) {
+    protected WebSocketClient(String timelyHostname, int timelyHttpsPort, int timelyWssPort, boolean clientAuth, boolean doLogin, String timelyUsername,
+                    String timelyPassword, String keyStoreFile, String keyStoreType, String keyStorePass, String trustStoreFile, String trustStoreType,
+                    String trustStorePass, boolean hostVerificationEnabled, int bufferSize) {
         this(HttpClient.getSSLContext(trustStoreFile, trustStoreType, trustStorePass, keyStoreFile, keyStoreType, keyStorePass), timelyHostname,
-                        timelyHttpsPort, timelyWssPort, clientAuth, doLogin, hostVerificationEnabled, bufferSize);
+                        timelyHttpsPort, timelyWssPort, clientAuth, doLogin, timelyUsername, timelyPassword, hostVerificationEnabled, bufferSize);
     }
 
     public void open(ClientHandler clientEndpoint) throws IOException, DeploymentException, URISyntaxException {
@@ -76,10 +92,23 @@ public class WebSocketClient implements AutoCloseable {
             try (CloseableHttpClient client = HttpClient.get(ssl, cookieJar, hostVerificationEnabled, clientAuth)) {
 
                 String target = "https://" + timelyHostname + ":" + timelyHttpsPort + "/login";
+
                 HttpRequestBase request = null;
-                // HTTP GET to /login to use certificate based login
-                request = new HttpGet(target);
-                log.trace("Performing client certificate login");
+                if (StringUtils.isEmpty(timelyUsername)) {
+                    // HTTP GET to /login to use certificate based login
+                    request = new HttpGet(target);
+                    LOG.trace("Performing client certificate login");
+                } else {
+                    // HTTP POST to /login to use username/password
+                    BasicAuthLogin login = new BasicAuthLogin();
+                    login.setUsername(timelyUsername);
+                    login.setPassword(timelyPassword);
+                    String payload = JsonSerializer.getObjectMapper().writeValueAsString(login);
+                    HttpPost post = new HttpPost(target);
+                    post.setEntity(new StringEntity(payload));
+                    request = post;
+                    LOG.trace("Performing BasicAuth login");
+                }
 
                 HttpResponse response = null;
                 try {
@@ -117,7 +146,7 @@ public class WebSocketClient implements AutoCloseable {
             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

@@ -18,30 +18,29 @@ 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);
+                    String timelyUsername, String timelyPassword, boolean hostVerificationEnabled, int bufferSize) {
+        super(ssl, timelyHostname, timelyHttpsPort, timelyWssPort, clientAuth, doLogin, timelyUsername, timelyPassword, hostVerificationEnabled, bufferSize);
         subscriptionId = UUID.randomUUID().toString();
-        log.trace("Created WebSocketClient with subscriptionId {}", this.subscriptionId);
+        LOG.trace("Created WebSocketClient with subscriptionId {}", this.subscriptionId);
     }
 
-    public WebSocketSubscriptionClient(String timelyHostname, int timelyHttpsPort, int timelyWssPort, boolean clientAuth, boolean doLogin, String keyStoreFile,
-                    String keyStoreType, String keyStorePass, String trustStoreFile, String trustStoreType, String trustStorePass,
-                    boolean hostVerificationEnabled, int bufferSize) {
-        super(timelyHostname, timelyHttpsPort, timelyWssPort, clientAuth, doLogin, keyStoreFile, keyStoreType, keyStorePass, trustStoreFile, trustStoreType,
-                        trustStorePass, hostVerificationEnabled, bufferSize);
+    public WebSocketSubscriptionClient(String timelyHostname, int timelyHttpsPort, int timelyWssPort, boolean clientAuth, boolean doLogin,
+                    String timelyUsername, String timelyPassword, String keyStoreFile, String keyStoreType, String keyStorePass, String trustStoreFile,
+                    String trustStoreType, String trustStorePass, boolean hostVerificationEnabled, int bufferSize) {
+        super(timelyHostname, timelyHttpsPort, timelyWssPort, clientAuth, doLogin, timelyUsername, timelyPassword, keyStoreFile, keyStoreType, keyStorePass,
+                        trustStoreFile, trustStoreType, trustStorePass, hostVerificationEnabled, bufferSize);
         subscriptionId = UUID.randomUUID().toString();
-        log.trace("Created WebSocketClient with subscriptionId {}", this.subscriptionId);
+        LOG.trace("Created WebSocketClient with subscriptionId {}", this.subscriptionId);
     }
 
     @Override
@@ -61,7 +60,7 @@ 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();
@@ -85,8 +84,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,8 +2,8 @@ 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;
@@ -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

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

@@ -48,7 +48,7 @@ public interface ObjectSizeOf {
 
     public static class Sizer {
 
-        private static final Logger log = LoggerFactory.getLogger(Sizer.class);
+        private static final Logger LOG = LoggerFactory.getLogger(Sizer.class);
         public static final short OBJECT_OVERHEAD = 8;
         public static final short ARRAY_OVERHEAD = 12;
         public static final short REFERENCE = 4;
@@ -148,7 +148,7 @@ public interface ObjectSizeOf {
                             }
                         }
                     } catch (Throwable t) {
-                        log.warn("Unable to determine object size for " + o, t);
+                        LOG.warn("Unable to determine object size for " + o, t);
                     }
                     if (debug) {
                         System.out.println(o.getClass().getCanonicalName() + "=" + size);

client/src/main/spotbugs/excludes.xml

@@ -1,5 +1,15 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <FindBugsFilter>
+  <Match>
+    <Bug pattern="DM_DEFAULT_ENCODING"/>
+    <Method name="connect" />
+    <Class name="timely.client.tcp.TcpClient"/>
+  </Match>
+  <Match>
+    <!-- Must ignore these everywhere, because of a javac byte code generation bug -->
+    <!-- https://github.com/spotbugs/spotbugs/issues/756 -->
+    <Bug pattern="RCN_REDUNDANT_NULLCHECK_WOULD_HAVE_BEEN_A_NPE" />
+  </Match>
   <Match>
     <!-- these are probably avoidable, but needs further triage to fix -->
     <Bug pattern="MS_EXPOSE_REP" />

client/src/test/java/timely/api/SerializationTest.java

@@ -11,6 +11,14 @@ public class SerializationTest {
 
     private ObjectMapper mapper = JsonSerializer.getObjectMapper();
 
+    @Test
+    public void testBasicAuth() throws Exception {
+        BasicAuthLogin login = new BasicAuthLogin();
+        login.setUsername("test");
+        login.setPassword("pass");
+        testSerialization(login);
+    }
+
     @Test
     public void testCreateSubscription() throws Exception {
         CreateSubscription create = new CreateSubscription();

collectd-nsq-plugin/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>2.0.2</version>
   </parent>
   <artifactId>collectd-nsq-plugin</artifactId>
   <dependencies>

collectd-nsq-plugin/src/main/java/timely/collectd/plugin/PooledCloseableHttpClientFactory.java

@@ -44,8 +44,8 @@ public class PooledCloseableHttpClientFactory implements PooledObjectFactory<Clo
 
     @Override
     public PooledObject<CloseableHttpClient> makeObject() throws Exception {
-        RequestConfig requestConfig = RequestConfig.custom().setConnectionRequestTimeout(connectionRequestTimeout).setConnectTimeout(connectTimeout)
-                        .setSocketTimeout(socketTimeout).build();
+        RequestConfig requestConfig = RequestConfig.custom().setConnectionRequestTimeout(connectionRequestTimeout)
+                .setConnectTimeout(connectTimeout).setSocketTimeout(socketTimeout).build();
         HttpClientBuilder builder = HttpClientBuilder.create();
         builder.setDefaultRequestConfig(requestConfig);
         builder.setConnectionTimeToLive(connectionTimeToLive, TimeUnit.MILLISECONDS);

collectd-nsq-plugin/src/main/java/timely/collectd/plugin/WriteNSQPlugin.java

@@ -23,11 +23,13 @@ import org.collectd.api.ValueList;
 
 /**
  *
- * CollectD plugin that writes metrics collected by the StatsD plugin to NSQ. NSQ can be configured to write statsd metrics to CollectD. If you do this, set the
- * statsd prefix in NSQ to 'nsq'.
+ * CollectD plugin that writes metrics collected by the StatsD plugin to NSQ.
+ * NSQ can be configured to write statsd metrics to CollectD. If you do this,
+ * set the statsd prefix in NSQ to 'nsq'.
  *
  */
-public class WriteNSQPlugin extends CollectDPluginParent implements CollectdConfigInterface, CollectdShutdownInterface, CollectdWriteInterface {
+public class WriteNSQPlugin extends CollectDPluginParent
+        implements CollectdConfigInterface, CollectdShutdownInterface, CollectdWriteInterface {
 
     private String host = null;
     private int port = 0;
@@ -38,7 +40,7 @@ public class WriteNSQPlugin extends CollectDPluginParent implements CollectdConf
 
     // pool should be limited by the number of WriteThreads configured in
     // collectd
-    static final private int POOL_MAX_SIZE = Integer.MAX_VALUE;
+    final private int POOL_MAX_SIZE = Integer.MAX_VALUE;
 
     public WriteNSQPlugin() {
         Collectd.registerConfig(WriteNSQPlugin.class.getName(), this);
@@ -76,10 +78,12 @@ public class WriteNSQPlugin extends CollectDPluginParent implements CollectdConf
             if (client != null) {
                 try {
                     client.close();
-                } catch (IOException e1) {}
+                } catch (IOException e1) {
+                }
                 try {
                     clientPool.invalidateObject(client);
-                } catch (Exception e2) {}
+                } catch (Exception e2) {
+                }
             }
         } finally {
             if (client != null) {
@@ -87,7 +91,8 @@ public class WriteNSQPlugin extends CollectDPluginParent implements CollectdConf
             }
             try {
                 out.close();
-            } catch (IOException e) {}
+            } catch (IOException e) {
+            }
         }
         return 0;
     }
@@ -100,7 +105,8 @@ public class WriteNSQPlugin extends CollectDPluginParent implements CollectdConf
             Collectd.logError("Error writing metric: " + e.getMessage());
             try {
                 out.close(); // Should cause IOException in write.
-            } catch (IOException e1) {}
+            } catch (IOException e1) {
+            }
         }
     }
 

collectd-plugin-parent/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>2.0.2</version>
   </parent>
   <artifactId>collectd-plugin-parent</artifactId>
   <dependencies>

collectd-plugin-parent/src/main/java/timely/collectd/plugin/CollectDPluginParent.java

@@ -15,7 +15,8 @@ import org.collectd.api.ValueList;
 public abstract class CollectDPluginParent {
 
     private static final String PUT = "put {0} {1} {2}{3}\n";
-    private static final Pattern HADOOP_STATSD_PATTERN = Pattern.compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)");
+    private static final Pattern HADOOP_STATSD_PATTERN = Pattern
+            .compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)");
     private static final String INSTANCE = " instance=";
     private static final String NAME = " name=";
     private static final String SAMPLE = " sample=";
@@ -28,7 +29,8 @@ public abstract class CollectDPluginParent {
     private static final String ABSOLUTE = "ABSOLUTE";
     private static final Pattern NSQ_PATTERN1 = Pattern.compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)");
     private static final Pattern NSQ_PATTERN2 = Pattern.compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)");
-    private static final Pattern NSQ_PATTERN3 = Pattern.compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_#]+)\\.([\\w-_]+)");
+    private static final Pattern NSQ_PATTERN3 = Pattern
+            .compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_#]+)\\.([\\w-_]+)");
     private static final Pattern ETHSTAT_PATTERN = Pattern.compile("([\\w-_]+)_queue_([\\w-_]+)_([\\w-_]+)");
     private static final String STATSD_PREFIX = "statsd";
     private static final String NSQ_PREFIX = STATSD_PREFIX + ".nsq.";
@@ -171,7 +173,8 @@ public abstract class CollectDPluginParent {
             metric.append("sys.snmp.").append(vl.getType());
             tags.append(INSTANCE).append(vl.getTypeInstance().replaceAll(" ", "_"));
         } else if (vl.getPlugin().equals("GenericJMX")) {
-            metric.append("sys.").append(vl.getPlugin()).append(PERIOD).append(vl.getType()).append(PERIOD).append(vl.getTypeInstance());
+            metric.append("sys.").append(vl.getPlugin()).append(PERIOD).append(vl.getType()).append(PERIOD)
+                    .append(vl.getTypeInstance());
             String[] pluginInstanceSplit = vl.getPluginInstance().split("-");
             if (pluginInstanceSplit.length > 0) {
                 tags.append(INSTANCE).append(pluginInstanceSplit[0].replaceAll(" ", "_"));
@@ -179,11 +182,14 @@ public abstract class CollectDPluginParent {
             if (pluginInstanceSplit.length > 1) {
                 tags.append(NAME).append(pluginInstanceSplit[1].replaceAll(" ", "_"));
             }
-        } else if (notEmpty(vl.getTypeInstance()) && notEmpty(vl.getType()) && notEmpty(vl.getPluginInstance()) && notEmpty(vl.getPlugin())) {
-            metric.append("sys.").append(vl.getPlugin()).append(PERIOD).append(vl.getType()).append(PERIOD).append(vl.getTypeInstance());
+        } else if (notEmpty(vl.getTypeInstance()) && notEmpty(vl.getType()) && notEmpty(vl.getPluginInstance())
+                && notEmpty(vl.getPlugin())) {
+            metric.append("sys.").append(vl.getPlugin()).append(PERIOD).append(vl.getType()).append(PERIOD)
+                    .append(vl.getTypeInstance());
             tags.append(INSTANCE).append(vl.getPluginInstance().replaceAll(" ", "_"));
         } else if (notEmpty(vl.getTypeInstance()) && notEmpty(vl.getType()) && notEmpty(vl.getPlugin())) {
-            metric.append("sys.").append(vl.getPlugin()).append(PERIOD).append(vl.getType()).append(PERIOD).append(vl.getTypeInstance());
+            metric.append("sys.").append(vl.getPlugin()).append(PERIOD).append(vl.getType()).append(PERIOD)
+                    .append(vl.getTypeInstance());
         } else if (notEmpty(vl.getType()) && notEmpty(vl.getPluginInstance()) && notEmpty(vl.getPlugin())) {
             metric.append("sys.").append(vl.getPlugin()).append(PERIOD).append(vl.getType());
             tags.append(INSTANCE).append(vl.getPluginInstance().replaceAll(" ", "_"));
@@ -206,7 +212,8 @@ public abstract class CollectDPluginParent {
                 tagsWithSample.append(SAMPLE_TYPE).append(sampleType);
             }
             Double value = vl.getValues().get(i).doubleValue();
-            String datapoint = MessageFormat.format(PUT, metricName, timestamp.toString(), value.toString(), tagsWithSample.toString());
+            String datapoint = MessageFormat.format(PUT, metricName, timestamp.toString(), value.toString(),
+                    tagsWithSample.toString());
             Collectd.logDebug("Writing: " + datapoint);
             write(datapoint, out);
         }

collectd-plugin-parent/src/main/java/timely/collectd/plugin/SMARTCodeMapping.java

@@ -6,13 +6,13 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
 
-public class SMARTCodeMapping extends HashMap<Integer,String> {
+public class SMARTCodeMapping extends HashMap<Integer, String> {
 
     private static final long serialVersionUID = 1L;
 
-    private static final Map<Integer,String> smart;
+    private static final Map<Integer, String> smart;
     static {
-        Map<Integer,String> map = new HashMap<>();
+        Map<Integer, String> map = new HashMap<>();
         /**
          * mapping from https://en.wikipedia.org/wiki/S.M.A.R.T.
          */
@@ -107,7 +107,8 @@ public class SMARTCodeMapping extends HashMap<Integer,String> {
         smart = Collections.unmodifiableMap(map);
     }
 
-    protected SMARTCodeMapping() {}
+    protected SMARTCodeMapping() {
+    }
 
     public int size() {
         return smart.size();
@@ -137,7 +138,7 @@ public class SMARTCodeMapping extends HashMap<Integer,String> {
         return smart.remove(key);
     }
 
-    public void putAll(Map<? extends Integer,? extends String> m) {
+    public void putAll(Map<? extends Integer, ? extends String> m) {
         smart.putAll(m);
     }
 
@@ -153,7 +154,7 @@ public class SMARTCodeMapping extends HashMap<Integer,String> {
         return smart.values();
     }
 
-    public Set<java.util.Map.Entry<Integer,String>> entrySet() {
+    public Set<java.util.Map.Entry<Integer, String>> entrySet() {
         return smart.entrySet();
     }
 

collectd-plugin-parent/src/test/java/timely/collectd/plugin/PluginParentTest.java

@@ -5,6 +5,7 @@ import static org.junit.Assert.*;
 import java.io.OutputStream;
 import java.util.Collections;
 
+import net.jcip.annotations.NotThreadSafe;
 import org.collectd.api.Collectd;
 import org.collectd.api.DataSet;
 import org.collectd.api.DataSource;
@@ -16,10 +17,8 @@ import org.powermock.api.easymock.PowerMock;
 import org.powermock.core.classloader.annotations.PrepareForTest;
 import org.powermock.modules.junit4.PowerMockRunner;
 
-import net.jcip.annotations.NotThreadSafe;
-
 @RunWith(PowerMockRunner.class)
-@PrepareForTest({Collectd.class})
+@PrepareForTest({ Collectd.class })
 @NotThreadSafe
 public class PluginParentTest {
 
@@ -60,7 +59,9 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put statsd.dfs.BlocksRead 1456156976840 1.0 host=r01n01 rack=r01 addl1=foo instance=DataNode sample=value sampleType=GAUGE\n", result);
+        assertEquals(
+                "put statsd.dfs.BlocksRead 1456156976840 1.0 host=r01n01 rack=r01 addl1=foo instance=DataNode sample=value sampleType=GAUGE\n",
+                result);
     }
 
     @Test
@@ -78,7 +79,8 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put statsd.baz 1456156976840 1.0 host=r01n01 rack=r01 addl1=foo sample=value sampleType=GAUGE\n", result);
+        assertEquals("put statsd.baz 1456156976840 1.0 host=r01n01 rack=r01 addl1=foo sample=value sampleType=GAUGE\n",
+                result);
     }
 
     @Test
@@ -96,7 +98,9 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put statsd.baz 1456156976840 1.0 host=r01n01 rack=r01 addl1=foo instance=bar sample=value sampleType=GAUGE\n", result);
+        assertEquals(
+                "put statsd.baz 1456156976840 1.0 host=r01n01 rack=r01 addl1=foo instance=bar sample=value sampleType=GAUGE\n",
+                result);
     }
 
     @Test
@@ -114,7 +118,9 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.hddtemp.temperature 1456156976840 35.0 host=r01n01 rack=r01 addl1=foo instance=sda sample=value sampleType=GAUGE\n", result);
+        assertEquals(
+                "put sys.hddtemp.temperature 1456156976840 35.0 host=r01n01 rack=r01 addl1=foo instance=sda sample=value sampleType=GAUGE\n",
+                result);
     }
 
     @Test
@@ -132,7 +138,9 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.smart.smart_badsectors 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo instance=sda sample=value sampleType=GAUGE\n", result);
+        assertEquals(
+                "put sys.smart.smart_badsectors 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo instance=sda sample=value sampleType=GAUGE\n",
+                result);
     }
 
     @Test
@@ -151,7 +159,9 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.smart.raw-read-error-rate 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo instance=sda sample=value sampleType=GAUGE\n", result);
+        assertEquals(
+                "put sys.smart.raw-read-error-rate 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo instance=sda sample=value sampleType=GAUGE\n",
+                result);
     }
 
     @Test
@@ -170,8 +180,9 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.smart.Total_LBAs_Read 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo code=242 instance=sda sample=value sampleType=GAUGE\n",
-                        result);
+        assertEquals(
+                "put sys.smart.Total_LBAs_Read 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo code=242 instance=sda sample=value sampleType=GAUGE\n",
+                result);
     }
 
     @Test
@@ -190,7 +201,9 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.snmp.if_octets 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo instance=Ethernet1 sample=value sampleType=GAUGE\n", result);
+        assertEquals(
+                "put sys.snmp.if_octets 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo instance=Ethernet1 sample=value sampleType=GAUGE\n",
+                result);
 
     }
 
@@ -210,8 +223,9 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.sensors.temperature.coretemp-isa-0000 1456156976840 35.0 host=r01n01 rack=r01 addl1=foo instance=1 sample=value sampleType=GAUGE\n",
-                        result);
+        assertEquals(
+                "put sys.sensors.temperature.coretemp-isa-0000 1456156976840 35.0 host=r01n01 rack=r01 addl1=foo instance=1 sample=value sampleType=GAUGE\n",
+                result);
     }
 
     @Test
@@ -229,7 +243,9 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.haproxy.run_queue 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo sample=value sampleType=GAUGE\n", result);
+        assertEquals(
+                "put sys.haproxy.run_queue 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo sample=value sampleType=GAUGE\n",
+                result);
     }
 
     @Test
@@ -248,8 +264,9 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.haproxy.queue_current 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo proxy_name=proxy1 service_name=server1 sample=value sampleType=GAUGE\n",
-                        result);
+        assertEquals(
+                "put sys.haproxy.queue_current 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo proxy_name=proxy1 service_name=server1 sample=value sampleType=GAUGE\n",
+                result);
     }
 
     @Test
@@ -268,8 +285,9 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.ethstat.tx_comp_queue_full 1456156976840 6.0 host=r01n01 rack=r01 addl1=foo instance=eth0 sample=value sampleType=GAUGE\n",
-                        result);
+        assertEquals(
+                "put sys.ethstat.tx_comp_queue_full 1456156976840 6.0 host=r01n01 rack=r01 addl1=foo instance=eth0 sample=value sampleType=GAUGE\n",
+                result);
 
     }
 
@@ -289,7 +307,9 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.ethstat.rx_bytes 1456156976840 6.0 host=r01n01 rack=r01 addl1=foo queue=1 instance=eth0 sample=value sampleType=GAUGE\n", result);
+        assertEquals(
+                "put sys.ethstat.rx_bytes 1456156976840 6.0 host=r01n01 rack=r01 addl1=foo queue=1 instance=eth0 sample=value sampleType=GAUGE\n",
+                result);
 
     }
 
@@ -308,8 +328,9 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.ipmi.temperature 1456156976840 6.0 host=r01n01 rack=r01 addl1=foo instance=LAN_NIC_Temp_system_board_(3.2) sample=value sampleType=GAUGE\n",
-                        result);
+        assertEquals(
+                "put sys.ipmi.temperature 1456156976840 6.0 host=r01n01 rack=r01 addl1=foo instance=LAN_NIC_Temp_system_board_(3.2) sample=value sampleType=GAUGE\n",
+                result);
 
     }
 

collectd-timely-plugin/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>2.0.2</version>
   </parent>
   <artifactId>collectd-timely-plugin</artifactId>
   <dependencies>

collectd-timely-plugin/src/main/java/timely/collectd/plugin/PooledSocketFactory.java

@@ -63,10 +63,11 @@ public class PooledSocketFactory implements PooledObjectFactory<Socket> {
                 if (socketTimeout > -1) {
                     socket.setSoTimeout(socketTimeout);
                 }
-                Collectd.logInfo("Connected to Timely at " + host + ":" + port + " from local port:" + socket.getLocalPort());
+                Collectd.logInfo(
+                        "Connected to Timely at " + host + ":" + port + " from local port:" + socket.getLocalPort());
             } catch (Exception e) {
-                Collectd.logError("Error connecting to Timely at " + host + ":" + port + ". Error: " + e.getMessage() + ".  Will retry connection in "
-                                + currentBackoff + " ms.");
+                Collectd.logError("Error connecting to Timely at " + host + ":" + port + ". Error: " + e.getMessage()
+                        + ".  Will retry connection in " + currentBackoff + " ms.");
                 try {
                     Thread.sleep(currentBackoff);
                 } catch (InterruptedException e1) {
@@ -93,7 +94,8 @@ public class PooledSocketFactory implements PooledObjectFactory<Socket> {
                 socket.close();
             }
         } catch (Exception e) {
-            Collectd.logError("Error closing connection to Timely at " + host + ":" + port + ". Error: " + e.getMessage());
+            Collectd.logError(
+                    "Error closing connection to Timely at " + host + ":" + port + ". Error: " + e.getMessage());
         } finally {
             pooledObject.invalidate();
             ;

collectd-timely-plugin/src/main/java/timely/collectd/plugin/WriteTimelyPlugin.java

@@ -21,13 +21,14 @@ import org.collectd.api.ValueList;
  * CollectD plugin that writes metrics collected by the StatsD plugin to Timely.
  *
  */
-public class WriteTimelyPlugin extends CollectDPluginParent implements CollectdConfigInterface, CollectdShutdownInterface, CollectdWriteInterface {
+public class WriteTimelyPlugin extends CollectDPluginParent
+        implements CollectdConfigInterface, CollectdShutdownInterface, CollectdWriteInterface {
 
     private GenericObjectPool<Socket> socketPool = null;
 
     // pool should be limited by the number of WriteThreads configured in
     // collectd
-    static final private int POOL_MAX_SIZE = Integer.MAX_VALUE;
+    final private int POOL_MAX_SIZE = Integer.MAX_VALUE;
 
     public WriteTimelyPlugin() {
         Collectd.registerConfig(WriteTimelyPlugin.class.getName(), this);

collectd-timely-plugin/src/test/java/timely/collectd/plugin/WriteTimelyPluginTest.java

@@ -7,6 +7,7 @@ import java.net.ServerSocket;
 import java.net.Socket;
 import java.util.Collections;
 
+import net.jcip.annotations.NotThreadSafe;
 import org.collectd.api.Collectd;
 import org.collectd.api.DataSet;
 import org.collectd.api.DataSource;
@@ -21,10 +22,8 @@ import org.powermock.core.classloader.annotations.PowerMockIgnore;
 import org.powermock.core.classloader.annotations.PrepareForTest;
 import org.powermock.modules.junit4.PowerMockRunner;
 
-import net.jcip.annotations.NotThreadSafe;
-
 @RunWith(PowerMockRunner.class)
-@PrepareForTest({Collectd.class})
+@PrepareForTest({ Collectd.class })
 @NotThreadSafe
 @PowerMockIgnore("javax.management.*")
 public class WriteTimelyPluginTest {

collectd-timely-plugin/src/test/resources/log4j2.xml

@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="WARN">
+  <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="TRACE">
+      <AppenderRef ref="CONSOLE"/>
+    </Root>
+  </Loggers>
+</Configuration>

collectd-timely-plugin/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

common/pom.xml

@@ -4,172 +4,14 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>2.0.2</version>
   </parent>
   <artifactId>timely-common</artifactId>
   <dependencies>
-    <dependency>
-      <groupId>com.fasterxml.jackson.core</groupId>
-      <artifactId>jackson-annotations</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>com.fasterxml.jackson.core</groupId>
-      <artifactId>jackson-core</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>com.fasterxml.jackson.core</groupId>
-      <artifactId>jackson-databind</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>com.fasterxml.jackson.dataformat</groupId>
-      <artifactId>jackson-dataformat-xml</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>com.fasterxml.jackson.dataformat</groupId>
-      <artifactId>jackson-dataformat-yaml</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>com.fasterxml.jackson.datatype</groupId>
-      <artifactId>jackson-datatype-guava</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>com.fasterxml.jackson.datatype</groupId>
-      <artifactId>jackson-datatype-jdk8</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>com.github.ben-manes.caffeine</groupId>
-      <artifactId>caffeine</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>com.google.flatbuffers</groupId>
-      <artifactId>flatbuffers-java</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>com.google.guava</groupId>
-      <artifactId>guava</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>commons-beanutils</groupId>
-      <artifactId>commons-beanutils</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>commons-lang</groupId>
-      <artifactId>commons-lang</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>gov.nsa.datawave</groupId>
-      <artifactId>datawave-in-memory-accumulo</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>io.github.classgraph</groupId>
-      <artifactId>classgraph</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>io.jsonwebtoken</groupId>
-      <artifactId>jjwt-impl</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>io.jsonwebtoken</groupId>
-      <artifactId>jjwt-jackson</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>io.netty</groupId>
-      <artifactId>netty-all</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>io.projectreactor.netty</groupId>
-      <artifactId>reactor-netty</artifactId>
-    </dependency>
-    <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-core</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.accumulo</groupId>
-      <artifactId>accumulo-minicluster</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.commons</groupId>
-      <artifactId>commons-lang3</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.commons</groupId>
-      <artifactId>commons-pool2</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.curator</groupId>
-      <artifactId>curator-framework</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.httpcomponents</groupId>
-      <artifactId>httpclient</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.springframework</groupId>
-      <artifactId>spring-webflux</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.springframework.boot</groupId>
-      <artifactId>spring-boot</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.springframework.boot</groupId>
-      <artifactId>spring-boot-starter</artifactId>
-      <exclusions>
-        <exclusion>
-          <groupId>org.springframework.boot</groupId>
-          <artifactId>spring-boot-starter-logging</artifactId>
-        </exclusion>
-      </exclusions>
-    </dependency>
-    <dependency>
-      <groupId>org.springframework.boot</groupId>
-      <artifactId>spring-boot-starter-validation</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.springframework.cloud</groupId>
-      <artifactId>spring-cloud-context</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.springframework.security</groupId>
-      <artifactId>spring-security-web</artifactId>
-    </dependency>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
   </dependencies>
-  <build>
-    <plugins>
-      <plugin>
-        <groupId>org.codehaus.mojo</groupId>
-        <artifactId>build-helper-maven-plugin</artifactId>
-        <executions>
-          <execution>
-            <id>add-source</id>
-            <goals>
-              <goal>add-source</goal>
-            </goals>
-            <phase>generate-sources</phase>
-            <configuration>
-              <sources>
-                <source>src/main/generated-java</source>
-              </sources>
-            </configuration>
-          </execution>
-        </executions>
-      </plugin>
-    </plugins>
-  </build>
 </project>

common/src/main/generated-java/timely/api/flatbuffer/Metrics.java

@@ -1,52 +0,0 @@
-// automatically generated by the FlatBuffers compiler, do not modify
-
-package timely.api.flatbuffer;
-
-import com.google.flatbuffers.BaseVector;
-import com.google.flatbuffers.Constants;
-import com.google.flatbuffers.FlatBufferBuilder;
-import com.google.flatbuffers.Table;
-import java.nio.ByteBuffer;
-import java.nio.ByteOrder;
-
-@SuppressWarnings("unused")
-public final class Metrics extends Table {
-  public static void ValidateVersion() { Constants.FLATBUFFERS_23_5_26(); }
-  public static Metrics getRootAsMetrics(ByteBuffer _bb) { return getRootAsMetrics(_bb, new Metrics()); }
-  public static Metrics getRootAsMetrics(ByteBuffer _bb, Metrics obj) { _bb.order(ByteOrder.LITTLE_ENDIAN); return (obj.__assign(_bb.getInt(_bb.position()) + _bb.position(), _bb)); }
-  public static boolean MetricsBufferHasIdentifier(ByteBuffer _bb) { return __has_identifier(_bb, "TMFB"); }
-  public void __init(int _i, ByteBuffer _bb) { __reset(_i, _bb); }
-  public Metrics __assign(int _i, ByteBuffer _bb) { __init(_i, _bb); return this; }
-
-  public timely.api.flatbuffer.Metric metrics(int j) { return metrics(new timely.api.flatbuffer.Metric(), j); }
-  public timely.api.flatbuffer.Metric metrics(timely.api.flatbuffer.Metric obj, int j) { int o = __offset(4); return o != 0 ? obj.__assign(__indirect(__vector(o) + j * 4), bb) : null; }
-  public int metricsLength() { int o = __offset(4); return o != 0 ? __vector_len(o) : 0; }
-  public timely.api.flatbuffer.Metric.Vector metricsVector() { return metricsVector(new timely.api.flatbuffer.Metric.Vector()); }
-  public timely.api.flatbuffer.Metric.Vector metricsVector(timely.api.flatbuffer.Metric.Vector obj) { int o = __offset(4); return o != 0 ? obj.__assign(__vector(o), 4, bb) : null; }
-
-  public static int createMetrics(FlatBufferBuilder builder,
-      int metricsOffset) {
-    builder.startTable(1);
-    Metrics.addMetrics(builder, metricsOffset);
-    return Metrics.endMetrics(builder);
-  }
-
-  public static void startMetrics(FlatBufferBuilder builder) { builder.startTable(1); }
-  public static void addMetrics(FlatBufferBuilder builder, int metricsOffset) { builder.addOffset(0, metricsOffset, 0); }
-  public static int createMetricsVector(FlatBufferBuilder builder, int[] data) { builder.startVector(4, data.length, 4); for (int i = data.length - 1; i >= 0; i--) builder.addOffset(data[i]); return builder.endVector(); }
-  public static void startMetricsVector(FlatBufferBuilder builder, int numElems) { builder.startVector(4, numElems, 4); }
-  public static int endMetrics(FlatBufferBuilder builder) {
-    int o = builder.endTable();
-    return o;
-  }
-  public static void finishMetricsBuffer(FlatBufferBuilder builder, int offset) { builder.finish(offset, "TMFB"); }
-  public static void finishSizePrefixedMetricsBuffer(FlatBufferBuilder builder, int offset) { builder.finishSizePrefixed(offset, "TMFB"); }
-
-  public static final class Vector extends BaseVector {
-    public Vector __assign(int _vector, int _element_size, ByteBuffer _bb) { __reset(_vector, _element_size, _bb); return this; }
-
-    public Metrics get(int j) { return get(new Metrics(), j); }
-    public Metrics get(Metrics obj, int j) {  return obj.__assign(__indirect(__element(j), bb), bb); }
-  }
-}
-

common/src/main/java/timely/Constants.java

@@ -1,8 +0,0 @@
-package timely;
-
-public class Constants {
-
-    public static final String SERVICE_DISCOVERY_PATH = "/timely/server/instances";
-    public static final String NON_CACHED_METRICS = "/timely/server/non_cached_metrics/data";
-    public static final String NON_CACHED_METRICS_LOCK_PATH = "/timely/server/non_cached_metrics/lock";
-}

common/src/main/java/timely/api/request/HttpGetRequest.java

@@ -1,14 +0,0 @@
-package timely.api.request;
-
-import io.netty.handler.codec.http.FullHttpRequest;
-import io.netty.handler.codec.http.QueryStringDecoder;
-
-public interface HttpGetRequest extends HttpRequest {
-
-    HttpGetRequest parseQueryParameters(QueryStringDecoder decoder) throws Exception;
-
-    void setHttpRequest(FullHttpRequest httpRequest);
-
-    FullHttpRequest getHttpRequest();
-
-}

common/src/main/java/timely/api/request/HttpPostRequest.java

@@ -1,7 +0,0 @@
-package timely.api.request;
-
-public interface HttpPostRequest extends HttpRequest {
-
-    HttpPostRequest parseBody(String content) throws Exception;
-
-}

common/src/main/java/timely/api/request/websocket/WebSocketRequest.java

@@ -1,8 +0,0 @@
-package timely.api.request.websocket;
-
-import com.fasterxml.jackson.annotation.JsonTypeInfo;
-
-import timely.api.request.Request;
-
-@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "operation")
-public interface WebSocketRequest extends Request {}

common/src/main/java/timely/auth/FileUserDetailsService.java

@@ -1,36 +0,0 @@
-package timely.auth;
-
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.springframework.security.core.userdetails.UserDetails;
-import org.springframework.security.core.userdetails.UsernameNotFoundException;
-import org.springframework.security.web.authentication.preauth.PreAuthenticatedAuthenticationToken;
-
-public class FileUserDetailsService
-                implements org.springframework.security.core.userdetails.AuthenticationUserDetailsService<PreAuthenticatedAuthenticationToken> {
-
-    private HashMap<String,TimelyUser> users;
-
-    public Map<String,TimelyUser> getUsers() {
-        return users;
-    }
-
-    public void setUsers(List<TimelyUser> users) {
-        this.users = new HashMap<>();
-        users.forEach(u -> this.users.put(u.getName(), u));
-    }
-
-    @Override
-    public UserDetails loadUserDetails(final PreAuthenticatedAuthenticationToken token) throws UsernameNotFoundException {
-        // Determine if the user is allowed to access the system, if not throw
-        // UsernameNotFoundException
-        String username = token.getName();
-        TimelyUser timelyUser = users.get(username);
-        if (timelyUser == null) {
-            throw new UsernameNotFoundException(username + " not configured.");
-        }
-        return new TimelyPrincipal(timelyUser);
-    }
-}

common/src/main/java/timely/auth/TimelyPrincipal.java

@@ -1,259 +0,0 @@
-package timely.auth;
-
-import static timely.auth.TimelyUser.ANONYMOUS_USER;
-
-import java.io.Serializable;
-import java.security.Principal;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.stream.Collectors;
-
-import javax.xml.bind.annotation.XmlAccessType;
-import javax.xml.bind.annotation.XmlAccessorType;
-import javax.xml.bind.annotation.XmlElement;
-import javax.xml.bind.annotation.XmlRootElement;
-import javax.xml.bind.annotation.XmlType;
-
-import org.springframework.security.core.GrantedAuthority;
-import org.springframework.security.core.authority.SimpleGrantedAuthority;
-import org.springframework.security.core.userdetails.UserDetails;
-
-import com.fasterxml.jackson.annotation.JsonIgnore;
-
-import timely.auth.TimelyUser.UserType;
-import timely.auth.util.DnUtils;
-
-/**
- * A {@link Principal} that represents a set of proxied {@link TimelyUser}s. For example, this proxied user could represent a GUI server acting on behalf of a
- * user. The GUI server user represents the entity that made the call to us and the other proxied user would be the actual end user.
- */
-@XmlRootElement
-@XmlType(factoryMethod = "anonymousPrincipal", propOrder = {"name", "proxiedUsers", "creationTime"})
-@XmlAccessorType(XmlAccessType.NONE)
-public class TimelyPrincipal implements Principal, Serializable, UserDetails {
-
-    private static final long serialVersionUID = 1L;
-    private final String username;
-    private final TimelyUser primaryUser;
-    @XmlElement
-    private final ArrayList<TimelyUser> proxiedUsers = new ArrayList<>();
-    @XmlElement
-    private final long creationTime;
-
-    /**
-     * This constructor should not be used. It is here to allow JAX-B mapping and CDI proxying of this class.
-     */
-    public TimelyPrincipal() {
-        this(ANONYMOUS_USER);
-    }
-
-    public TimelyPrincipal(String username) {
-        this.creationTime = System.currentTimeMillis();
-        this.primaryUser = ANONYMOUS_USER;
-        this.proxiedUsers.add(ANONYMOUS_USER);
-        this.username = username;
-    }
-
-    public TimelyPrincipal(TimelyUser timelyUser) {
-        this.creationTime = System.currentTimeMillis();
-        this.primaryUser = timelyUser;
-        this.proxiedUsers.add(timelyUser);
-        this.username = timelyUser.getName();
-    }
-
-    public TimelyPrincipal(Collection<TimelyUser> proxiedUsers) {
-        this(proxiedUsers, System.currentTimeMillis());
-    }
-
-    public TimelyPrincipal(Collection<? extends TimelyUser> proxiedUsers, long creationTime) {
-        this.proxiedUsers.addAll(proxiedUsers);
-        this.creationTime = creationTime;
-        this.primaryUser = TimelyPrincipal.findPrimaryUser(this.proxiedUsers);
-        this.username = TimelyPrincipal.orderProxiedUsers(this.proxiedUsers).stream().map(TimelyUser::getName).collect(Collectors.joining(" -> "));
-    }
-
-    /**
-     * Gets the {@link TimelyUser} that represents the primary user in this TimelyPrincipal. If there is only one TimelyUser, then it is the primaryUser. If
-     * there is more than one TimelyUser, then the first (and presumably only) TimelyUser whose {@link TimelyUser#getUserType()} is {@link UserType#USER} is the
-     * primary user. If no such TimelyUser is present, then the first principal in the list is returned as the primary user. This will be the first entity in
-     * the X-ProxiedEntitiesChain which should be the server that originated the request.
-     */
-    static protected TimelyUser findPrimaryUser(List<TimelyUser> timelyUsers) {
-        if (timelyUsers.isEmpty()) {
-            return null;
-        } else {
-            return timelyUsers.get(findPrimaryUserPosition(timelyUsers));
-        }
-    }
-
-    static protected int findPrimaryUserPosition(List<TimelyUser> datawaveUsers) {
-        if (datawaveUsers.isEmpty()) {
-            return -1;
-        } else {
-            for (int x = 0; x < datawaveUsers.size(); x++) {
-                if (datawaveUsers.get(x).getUserType().equals(UserType.USER)) {
-                    return x;
-                }
-            }
-            return 0;
-        }
-    }
-
-    /*
-     * The purpose here is to return a List of TimelyUsers where the original caller is first followed by any entities in X-ProxiedEntitiesChain in the order
-     * that they were traversed and ending with the entity that made the final call. The List that is passed is not modified. This method makes the following
-     * assumptions about the List that is passed to ths method: 1) The first element is the one that made the final call 2) Additional elements (if any) are
-     * from X-ProxiedEntitiesChain in chronological order of the calls
-     */
-    static protected List<TimelyUser> orderProxiedUsers(List<TimelyUser> timelyUsers) {
-        List<TimelyUser> users = new ArrayList<>();
-        int position = TimelyPrincipal.findPrimaryUserPosition(timelyUsers);
-        if (position >= 0) {
-            users.add(timelyUsers.get(position));
-            if (timelyUsers.size() > 1) {
-                timelyUsers.stream().limit(position).forEach(u -> users.add(u));
-                timelyUsers.stream().skip(position + 1).forEach(u -> users.add(u));
-            }
-        }
-        return users;
-    }
-
-    public Collection<TimelyUser> getProxiedUsers() {
-        return Collections.unmodifiableCollection(this.proxiedUsers);
-    }
-
-    public TimelyUser getPrimaryUser() {
-        return primaryUser;
-    }
-
-    public Collection<? extends Collection<String>> getAuthorizations() {
-        // @formatter:off
-        return Collections.unmodifiableCollection(
-                TimelyPrincipal.orderProxiedUsers(this.proxiedUsers).stream()
-                        .map(TimelyUser::getAuths)
-                        .collect(Collectors.toList()));
-        // @formatter:on
-    }
-
-    public String[] getDNs() {
-        // @formatter:off
-        return TimelyPrincipal.orderProxiedUsers(this.proxiedUsers).stream()
-                .map(TimelyUser::getDn)
-                .map(SubjectIssuerDNPair::subjectDN)
-                .toArray(String[]::new);
-        // @formatter:on
-    }
-
-    public long getCreationTime() {
-        return this.creationTime;
-    }
-
-    @Override
-    @XmlElement
-    public String getName() {
-        return this.username;
-    }
-
-    public String getShortName() {
-        return DnUtils.getShortName(getPrimaryUser().getName());
-    }
-
-    public SubjectIssuerDNPair getUserDN() {
-        return getPrimaryUser().getDn();
-    }
-
-    public List<String> getProxyServers() {
-
-        // @formatter:off
-        List<String> proxyServers = orderProxiedUsers(this.proxiedUsers).stream()
-                .filter(u -> u.getUserType() == UserType.SERVER)
-                .filter(u -> !u.equals(this.primaryUser))
-                .map(TimelyUser::getDn)
-                .map(SubjectIssuerDNPair::subjectDN)
-                .collect(Collectors.toList());
-        // @formatter:on
-        return proxyServers.isEmpty() ? null : proxyServers;
-    }
-
-    @Override
-    public boolean equals(Object o) {
-        if (this == o)
-            return true;
-        if (o == null || getClass() != o.getClass())
-            return false;
-
-        TimelyPrincipal that = (TimelyPrincipal) o;
-
-        if (!username.equals(that.username))
-            return false;
-        return proxiedUsers.equals(that.proxiedUsers);
-    }
-
-    @Override
-    public int hashCode() {
-        int result = username.hashCode();
-        result = 31 * result + proxiedUsers.hashCode();
-        return result;
-    }
-
-    @Override
-    public String toString() {
-        return "TimelyPrincipal{" + "name='" + username + "'" + ", proxiedUsers=" + TimelyPrincipal.orderProxiedUsers(proxiedUsers) + "}";
-    }
-
-    public static TimelyPrincipal anonymousPrincipal() {
-        return new TimelyPrincipal("ANONYMOUS");
-    }
-
-    public String getAuthorizationsString() {
-        return orderProxiedUsers(this.proxiedUsers).stream().map(TimelyUser::getAuths).map(a -> a.toString()).collect(Collectors.joining(" -> ", "[", "]"));
-    }
-
-    @Override
-    @JsonIgnore
-    public Collection<? extends GrantedAuthority> getAuthorities() {
-        final Collection<SimpleGrantedAuthority> auths = new ArrayList<>();
-        for (String auth : getPrimaryUser().getAuths()) {
-            auths.add(new SimpleGrantedAuthority(auth));
-        }
-        return auths;
-    }
-
-    @Override
-    @JsonIgnore
-    public String getPassword() {
-        return "";
-    }
-
-    @Override
-    @JsonIgnore
-    public String getUsername() {
-        return getPrimaryUser().getName();
-    }
-
-    @Override
-    @JsonIgnore
-    public boolean isAccountNonExpired() {
-        return true;
-    }
-
-    @Override
-    @JsonIgnore
-    public boolean isAccountNonLocked() {
-        return true;
-    }
-
-    @Override
-    @JsonIgnore
-    public boolean isCredentialsNonExpired() {
-        return true;
-    }
-
-    @Override
-    @JsonIgnore
-    public boolean isEnabled() {
-        return true;
-    }
-}