Fix deadlock in tracking noncachedMetrics across processes

balancer/bin/timely-balancer.sh

@@ -8,11 +8,6 @@ else
         TCNATIVE_SUFFIX="so"
 fi
 
-# netty tcnative file reference
-#netty-tcnative-1.1.33.Fork18-linux-x86_64-fedora.jar -> libnetty-tcnative.so
-#netty-tcnative-1.1.33.Fork18-linux-x86_64.jar -> libnetty-tcnative.so
-#netty-tcnative-1.1.33.Fork18-osx-x86_64.jar -> libnetty-tcnative.jnilib
-
 THIS_DIR="${THIS_SCRIPT%/*}"
 NATIVE_DIR="${THIS_DIR}/META-INF/native"
 BASE_DIR=${THIS_DIR}/..
@@ -32,14 +27,16 @@ fi
 mkdir -p ${NATIVE_DIR}
 
 pushd ${BASE_DIR}/bin
-$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative*.jar META-INF/native/libnetty_tcnative.${NATIVE_SUFFIX}
+$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative*.jar META-INF/native/libnetty_tcnative.${TCNATIVE_SUFFIX}
-$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-transport-native-epoll*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${NATIVE_SUFFIX}
+$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-transport-native-epoll*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
 popd
 
-export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*"
+export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*:${HADOOP_CONF_DIR}"
 JVM_ARGS="-Xmx1G -Xms1G -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS}"
 JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
 JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
+JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}/libnetty-${TCNATIVE_SUFFIX}"
+
 #JVM_ARGS="${JVM_ARGS} -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=5005"
 
 echo "$JAVA_HOME/bin/java ${JVM_ARGS} timely.balancer.Balancer --spring.config.name=timely --spring.profiles.active=balancer"

balancer/conf/log4j2-spring.xml

@@ -22,5 +22,8 @@
       <AppenderRef ref="Console" level="warn" />
       <AppenderRef ref="Rolling" level="trace" />
     </Root>
+    <Logger name="timely" additivity="false">
+      <AppenderRef ref="Console" level="debug" />
+    </Logger>
   </Loggers>
 </Configuration>

balancer/conf/timely-balancer.yml

@@ -13,7 +13,7 @@ timely-balancer:
       certificate-file:
       key-file:
       #key-password:
-      use-generated-keypair: false
+      use-generated-keypair: true
       trust-store-file:
       use-openssl: true
       #use-ciphers:

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

@@ -1,5 +1,9 @@
 package timely.balancer;
 
+import static timely.Server.SERVICE_DISCOVERY_PATH;
+import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS;
+import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS_LOCK_PATH;
+
 import java.io.File;
 import java.net.InetSocketAddress;
 import java.nio.charset.StandardCharsets;
@@ -52,6 +56,12 @@ import io.netty.handler.stream.ChunkedWriteHandler;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.util.NettyRuntime;
 import io.netty.util.internal.SystemPropertyUtil;
+import org.apache.curator.RetryPolicy;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryForever;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.data.Stat;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.boot.builder.SpringApplicationBuilder;
@@ -115,9 +125,13 @@ public class Balancer {
     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));
@@ -239,6 +253,10 @@ public class Balancer {
             LOG.info("Closing applicationContext");
             applicationContext.close();
         }
+
+        if (curatorFramework != null) {
+            curatorFramework.close();
+        }
         this.shutdown = true;
         LOG.info("Server shut down.");
     }
@@ -320,8 +338,30 @@ public class Balancer {
                 ssl.getKeyFile(), ssl.getKeyType(), ssl.getKeyPassword());
     }
 
+    private void ensureZkPaths(CuratorFramework curatorFramework, String[] paths) {
+        for (String s : paths) {
+            try {
+                Stat stat = curatorFramework.checkExists().forPath(s);
+                if (stat == null) {
+                    curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT)
+                            .forPath(s);
+                }
+            } catch (Exception e) {
+                LOG.info(e.getMessage());
+            }
+
+        }
+    }
+
     public void run() throws Exception {
 
+        // start curator framework
+        RetryPolicy retryPolicy = new RetryForever(1000);
+        curatorFramework = CuratorFrameworkFactory.newClient(balancerConfig.getZooKeeper().getServers(), 30000, 1000,
+                retryPolicy);
+        curatorFramework.start();
+        ensureZkPaths(curatorFramework, zkPaths);
+
         final boolean useEpoll = useEpoll();
         Class<? extends ServerSocketChannel> channelClass;
         Class<? extends Channel> datagramChannelClass;
@@ -355,7 +395,7 @@ public class Balancer {
             tcpClientPools.add(new TcpClientPool(this.balancerConfig));
         }
         HealthChecker healthChecker = new HealthChecker(this.balancerConfig, tcpClientPools.get(0));
-        this.metricResolver = new BalancedMetricResolver(this.balancerConfig, healthChecker);
+        this.metricResolver = new BalancedMetricResolver(curatorFramework, this.balancerConfig, healthChecker);
 
         final ServerBootstrap tcpServer = new ServerBootstrap();
         tcpServer.group(tcpBossGroup, tcpWorkerGroup);

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

@@ -1,5 +1,7 @@
 package timely.balancer.connection;
 
+import java.util.Timer;
+
 import org.apache.commons.lang3.builder.EqualsBuilder;
 import org.apache.commons.lang3.builder.HashCodeBuilder;
 import org.apache.commons.lang3.builder.ToStringBuilder;
@@ -8,8 +10,6 @@ import org.slf4j.LoggerFactory;
 import timely.balancer.ArrivalRate;
 import timely.balancer.configuration.BalancerConfiguration;
 
-import java.util.Timer;
-
 public class TimelyBalancedHost {
 
     private static final Logger LOG = LoggerFactory.getLogger(TimelyBalancedHost.class);

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

@@ -4,6 +4,8 @@ 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.nio.charset.Charset;
 import java.util.ArrayList;
@@ -20,6 +22,8 @@ import java.util.Set;
 import java.util.Timer;
 import java.util.TimerTask;
 import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -27,10 +31,10 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
 import com.csvreader.CsvReader;
 import com.csvreader.CsvWriter;
 import org.apache.commons.lang.StringUtils;
-import org.apache.curator.RetryPolicy;
+import org.apache.commons.lang3.SerializationUtils;
 import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
+import org.apache.curator.framework.recipes.atomic.DistributedAtomicValue;
 import org.apache.curator.framework.recipes.cache.TreeCache;
 import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
 import org.apache.curator.framework.recipes.cache.TreeCacheListener;
@@ -38,7 +42,7 @@ import org.apache.curator.framework.recipes.leader.LeaderLatch;
 import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
 import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
 import org.apache.curator.framework.state.ConnectionState;
-import org.apache.curator.retry.RetryUntilElapsed;
+import org.apache.curator.retry.RetryForever;
 import org.apache.curator.x.discovery.ServiceCache;
 import org.apache.curator.x.discovery.ServiceDiscovery;
 import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
@@ -49,7 +53,7 @@ import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
-import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.CreateMode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import timely.ServerDetails;
@@ -71,43 +75,37 @@ public class BalancedMetricResolver implements MetricResolver {
     private Timer arrivalRateTimer = new Timer("AriivalRateTimerResolver", true);
     private int roundRobinCounter = 0;
     private Set<String> nonCachedMetrics = new HashSet<>();
-    private ReentrantReadWriteLock nonCachedMetricsLock = new ReentrantReadWriteLock();
+    private DistributedAtomicValue nonCachedMetricsIP;
+    private ReentrantReadWriteLock nonCachedMetricsLocalLock = new ReentrantReadWriteLock();
+    private InterProcessReadWriteLock nonCachedMetricsIPRWLock;
     private BalancerConfiguration balancerConfig;
-    private CuratorFramework curatorFramework;
-    private RetryPolicy retryPolicy = new RetryUntilElapsed(60000, 1000);
     private LeaderLatch leaderLatch;
     private AtomicBoolean isLeader = new AtomicBoolean(false);
-    private InterProcessReadWriteLock assignmentsLock;
+    private InterProcessReadWriteLock assignmentsIPRWLock;
     private DistributedAtomicLong assignmentsLastUpdatedInHdfs;
     private AtomicLong assignmentsLastUpdatedLocal = new AtomicLong(0);
 
-    private String[] zkPaths = new String[] { LEADER_LATCH_PATH, ASSIGNMENTS_LAST_UPDATED_PATH, ASSIGNMENTS_LOCK_PATH,
-            SERVICE_DISCOVERY_PATH };
-
     private enum BalanceType {
-        HIGH_LOW, HIGH_AVG, AVG_LOW;
+        HIGH_LOW, HIGH_AVG, AVG_LOW
     }
 
-    public BalancedMetricResolver(BalancerConfiguration balancerConfig, HealthChecker healthChecker) {
+    public BalancedMetricResolver(CuratorFramework curatorFramework, BalancerConfiguration balancerConfig,
+            HealthChecker healthChecker) {
         this.balancerConfig = balancerConfig;
         this.healthChecker = healthChecker;
 
-        // start curator framework
+        assignmentsIPRWLock = new InterProcessReadWriteLock(curatorFramework, ASSIGNMENTS_LOCK_PATH);
-        curatorFramework = CuratorFrameworkFactory.newClient(balancerConfig.getZooKeeper().getServers(), 30000, 1000,
+        testIPRWLock(curatorFramework, assignmentsIPRWLock, ASSIGNMENTS_LOCK_PATH);
-                retryPolicy);
-        curatorFramework.start();
-        ensureZkPaths(curatorFramework, zkPaths);
-        assignmentsLock = new InterProcessReadWriteLock(curatorFramework, ASSIGNMENTS_LOCK_PATH);
         startLeaderLatch(curatorFramework);
         startServiceListener(curatorFramework);
         assignmentsLastUpdatedInHdfs = new DistributedAtomicLong(curatorFramework, ASSIGNMENTS_LAST_UPDATED_PATH,
-                retryPolicy);
+                new RetryForever(1000));
 
-        TreeCacheListener listener = new TreeCacheListener() {
+        TreeCacheListener assignmentListener = new TreeCacheListener() {
 
             @Override
             public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent event) throws Exception {
-                LOG.info("Handling event {}. assignmentsLastUpdatedInHdfs:{}", event.getType().toString(),
+                LOG.info("Handling assignments event {}. assignmentsLastUpdatedInHdfs:{}", event.getType().toString(),
                         new Date(assignmentsLastUpdatedInHdfs.get().postValue()));
                 if (event.getType().equals(TreeCacheEvent.Type.NODE_UPDATED)) {
                     long lastLocalUpdate = assignmentsLastUpdatedLocal.get();
@@ -125,19 +123,42 @@ public class BalancedMetricResolver implements MetricResolver {
         };
 
         try {
-            TreeCache treeCache = new TreeCache(curatorFramework, ASSIGNMENTS_LAST_UPDATED_PATH);
+            TreeCache assignmentTreeCache = new TreeCache(curatorFramework, ASSIGNMENTS_LAST_UPDATED_PATH);
-            treeCache.getListenable().addListener(listener);
+            assignmentTreeCache.getListenable().addListener(assignmentListener);
-            treeCache.start();
+            assignmentTreeCache.start();
         } catch (Exception e) {
             LOG.error(e.getMessage(), e);
         }
 
-        nonCachedMetricsLock.writeLock().lock();
+        nonCachedMetricsIPRWLock = new InterProcessReadWriteLock(curatorFramework, NON_CACHED_METRICS_LOCK_PATH);
-        try {
+        testIPRWLock(curatorFramework, nonCachedMetricsIPRWLock, NON_CACHED_METRICS_LOCK_PATH);
-            nonCachedMetrics.addAll(balancerConfig.getCache().getNonCachedMetrics());
+        nonCachedMetricsIP = new DistributedAtomicValue(curatorFramework, NON_CACHED_METRICS, new RetryForever(1000));
-        } finally {
+        TreeCacheListener nonCachedMetricsListener = new TreeCacheListener() {
-            nonCachedMetricsLock.writeLock().unlock();
+
+            @Override
+            public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent event) throws Exception {
+                if (event.getType().equals(TreeCacheEvent.Type.NODE_UPDATED)) {
+                    LOG.info("Handling nonCachedMetricsIP event {}", event.getType().toString());
+                    readNonCachedMetricsIP();
                 }
+            }
+        };
+
+        try {
+            TreeCache nonCachedMetricsTreeCache = new TreeCache(curatorFramework, NON_CACHED_METRICS);
+            nonCachedMetricsTreeCache.getListenable().addListener(nonCachedMetricsListener);
+            nonCachedMetricsTreeCache.start();
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+        }
+
+        nonCachedMetricsLocalLock.writeLock().lock();
+        try {
+            addNonCachedMetrics(balancerConfig.getCache().getNonCachedMetrics());
+        } finally {
+            nonCachedMetricsLocalLock.writeLock().unlock();
+        }
+        readNonCachedMetricsIP();
 
         readAssignmentsFromHdfs();
 
@@ -176,17 +197,53 @@ public class BalancedMetricResolver implements MetricResolver {
         }, 10000, 60000);
     }
 
-    private void ensureZkPaths(CuratorFramework curatorFramework, String[] paths) {
+    private void readNonCachedMetricsIP() {
-        for (String s : paths) {
         try {
-                Stat stat = curatorFramework.checkExists().forPath(s);
+            nonCachedMetricsLocalLock.readLock().lock();
-                if (stat == null) {
+            Set<String> currentNonCachedMetricsDistributed;
-                    curatorFramework.create().creatingParentContainersIfNeeded().forPath(s);
+            try {
-                }
+                nonCachedMetricsIPRWLock.readLock().acquire();
+                byte[] currentNonCachedMetricsDistributedBytes = nonCachedMetricsIP.get().postValue();
+                if (currentNonCachedMetricsDistributedBytes == null) {
+                    currentNonCachedMetricsDistributed = new TreeSet<>();
+                } else {
+                    try {
+                        currentNonCachedMetricsDistributed = SerializationUtils
+                                .deserialize(currentNonCachedMetricsDistributedBytes);
                     } catch (Exception e) {
-                LOG.info(e.getMessage());
+                        LOG.error(e.getMessage());
+                        currentNonCachedMetricsDistributed = new TreeSet<>();
+                    }
+                }
+            } finally {
+                nonCachedMetricsIPRWLock.readLock().release();
             }
 
+            if (nonCachedMetrics.containsAll(currentNonCachedMetricsDistributed)) {
+                LOG.info("local nonCachedMetrics already contains {}", currentNonCachedMetricsDistributed);
+            } else {
+                nonCachedMetricsLocalLock.readLock().unlock();
+                nonCachedMetricsLocalLock.writeLock().lock();
+                currentNonCachedMetricsDistributed.removeAll(nonCachedMetrics);
+                LOG.info("Adding {} to local nonCachedMetrics", currentNonCachedMetricsDistributed);
+                nonCachedMetrics.addAll(currentNonCachedMetricsDistributed);
+                balancerLock.writeLock().lock();
+                try {
+                    for (String m : currentNonCachedMetricsDistributed) {
+                        metricToHostMap.remove(m);
+                    }
+                } finally {
+                    balancerLock.writeLock().unlock();
+                }
+            }
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+        } finally {
+            if (nonCachedMetricsLocalLock.isWriteLockedByCurrentThread()) {
+                nonCachedMetricsLocalLock.writeLock().unlock();
+            } else {
+                nonCachedMetricsLocalLock.readLock().unlock();
+            }
         }
     }
 
@@ -582,6 +639,78 @@ public class BalancedMetricResolver implements MetricResolver {
         return numReassigned;
     }
 
+    private void addNonCachedMetrics(Collection<String> nonCachedMetricsUpdate) {
+        if (!nonCachedMetricsUpdate.isEmpty()) {
+            try {
+                LOG.info("Adding {} to local nonCachedMetrics", nonCachedMetricsUpdate);
+                nonCachedMetricsLocalLock.writeLock().lock();
+                try {
+                    nonCachedMetrics.addAll(nonCachedMetricsUpdate);
+                } finally {
+                    nonCachedMetricsLocalLock.writeLock().unlock();
+                }
+                balancerLock.writeLock().lock();
+                try {
+                    for (String m : nonCachedMetricsUpdate) {
+                        metricToHostMap.remove(m);
+                    }
+                } finally {
+                    balancerLock.writeLock().unlock();
+                }
+
+                try {
+                    nonCachedMetricsIPRWLock.writeLock().acquire();
+                    byte[] currentNonCachedMetricsDistributedBytes = nonCachedMetricsIP.get().postValue();
+                    Set<String> currentNonCachedMetricsIP;
+                    if (currentNonCachedMetricsDistributedBytes == null) {
+                        currentNonCachedMetricsIP = new TreeSet<>();
+                    } else {
+                        currentNonCachedMetricsIP = SerializationUtils
+                                .deserialize(currentNonCachedMetricsDistributedBytes);
+                    }
+                    if (currentNonCachedMetricsIP.containsAll(nonCachedMetricsUpdate)) {
+                        LOG.info("nonCachedMetricsIP already contains {}", nonCachedMetricsUpdate);
+                    } else {
+                        nonCachedMetricsUpdate.removeAll(currentNonCachedMetricsIP);
+                        LOG.info("Adding {} to nonCachedMetricsIP", nonCachedMetricsUpdate);
+                        TreeSet<String> updateSet = new TreeSet<>();
+                        updateSet.addAll(currentNonCachedMetricsIP);
+                        updateSet.addAll(nonCachedMetricsUpdate);
+                        byte[] updateValue = SerializationUtils.serialize(updateSet);
+                        nonCachedMetricsIP.trySet(updateValue);
+                        if (!nonCachedMetricsIP.get().succeeded()) {
+                            nonCachedMetricsIP.forceSet(updateValue);
+                        }
+                    }
+                } finally {
+                    nonCachedMetricsIPRWLock.writeLock().release();
+                }
+            } catch (Exception e) {
+                LOG.error(e.getMessage(), e);
+            }
+        }
+    }
+
+    private void testIPRWLock(CuratorFramework curatorFramework, InterProcessReadWriteLock lock, String path) {
+        try {
+            lock.writeLock().acquire(10, TimeUnit.SECONDS);
+        } catch (Exception e1) {
+            try {
+                curatorFramework.delete().deletingChildrenIfNeeded().forPath(path);
+                curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT)
+                        .forPath(path);
+            } catch (Exception e2) {
+                LOG.info(e2.getMessage(), e2);
+            }
+        } finally {
+            try {
+                lock.writeLock().release();
+            } catch (Exception e3) {
+                LOG.error(e3.getMessage());
+            }
+        }
+    }
+
     private boolean shouldCache(String metricName) {
 
         if (StringUtils.isBlank(metricName)) {
@@ -589,6 +718,14 @@ public class BalancedMetricResolver implements MetricResolver {
         } else {
             balancerLock.readLock().lock();
             try {
+                nonCachedMetricsLocalLock.readLock().lock();
+                try {
+                    if (nonCachedMetrics.contains(metricName)) {
+                        return false;
+                    }
+                } finally {
+                    nonCachedMetricsLocalLock.readLock().unlock();
+                }
                 if (metricToHostMap.containsKey(metricName)) {
                     return true;
                 }
@@ -596,27 +733,17 @@ public class BalancedMetricResolver implements MetricResolver {
                 balancerLock.readLock().unlock();
             }
 
-            nonCachedMetricsLock.readLock().lock();
+            nonCachedMetricsLocalLock.writeLock().lock();
             try {
-                if (nonCachedMetrics.contains(metricName)) {
-                    return false;
-                }
-
                 for (String r : nonCachedMetrics) {
                     if (metricName.matches(r)) {
                         LOG.info("Adding {} to list of non-cached metrics", metricName);
-                        nonCachedMetricsLock.readLock().unlock();
+                        addNonCachedMetrics(Collections.singleton(metricName));
-                        nonCachedMetricsLock.writeLock().lock();
-                        nonCachedMetrics.add(metricName);
                         return false;
                     }
                 }
             } finally {
-                if (nonCachedMetricsLock.isWriteLockedByCurrentThread()) {
+                nonCachedMetricsLocalLock.writeLock().unlock();
-                    nonCachedMetricsLock.writeLock().unlock();
-                } else {
-                    nonCachedMetricsLock.readLock().unlock();
-                }
             }
             return true;
         }
@@ -768,7 +895,8 @@ public class BalancedMetricResolver implements MetricResolver {
 
         Map<String, TimelyBalancedHost> assignedMetricToHostMap = new TreeMap<>();
         try {
-            assignmentsLock.readLock().acquire();
+            assignmentsIPRWLock.readLock().acquire();
+            balancerLock.writeLock().lock();
             Configuration configuration = new Configuration();
             FileSystem fs = FileSystem.get(configuration);
             Path assignmentFile = new Path(balancerConfig.getAssignmentFile());
@@ -798,16 +926,24 @@ public class BalancedMetricResolver implements MetricResolver {
                 }
             }
 
-            balancerLock.writeLock().lock();
-            try {
             metricToHostMap.clear();
             for (Map.Entry<String, TimelyBalancedHost> e : assignedMetricToHostMap.entrySet()) {
                 if (shouldCache(e.getKey())) {
                     metricToHostMap.put(e.getKey(), e.getValue());
                 }
             }
+
+            nonCachedMetricsLocalLock.readLock().lock();
+            try {
+                Iterator<Map.Entry<String, TimelyBalancedHost>> itr = metricToHostMap.entrySet().iterator();
+                while (itr.hasNext()) {
+                    Map.Entry<String, TimelyBalancedHost> e = itr.next();
+                    if (nonCachedMetrics.contains(e.getKey())) {
+                        itr.remove();
+                    }
+                }
             } finally {
-                balancerLock.writeLock().unlock();
+                nonCachedMetricsLocalLock.readLock().unlock();
             }
             assignmentsLastUpdatedLocal.set(assignmentsLastUpdatedInHdfs.get().postValue());
             LOG.info("Read {} assignments from hdfs lastHdfsUpdate = lastLocalUpdate ({})", metricToHostMap.size(),
@@ -815,8 +951,9 @@ public class BalancedMetricResolver implements MetricResolver {
         } catch (Exception e) {
             LOG.error(e.getMessage(), e);
         } finally {
+            balancerLock.writeLock().unlock();
             try {
-                assignmentsLock.readLock().release();
+                assignmentsIPRWLock.readLock().release();
             } catch (Exception e) {
                 LOG.error(e.getMessage(), e);
             }
@@ -827,8 +964,9 @@ public class BalancedMetricResolver implements MetricResolver {
 
         CsvWriter writer = null;
         try {
-            assignmentsLock.writeLock().acquire();
+            assignmentsIPRWLock.writeLock().acquire();
             balancerLock.readLock().lock();
+            nonCachedMetricsLocalLock.readLock().lock();
             try {
                 if (!metricToHostMap.isEmpty()) {
                     Configuration configuration = new Configuration();
@@ -845,6 +983,7 @@ public class BalancedMetricResolver implements MetricResolver {
                     writer.write("tcpPort");
                     writer.endRecord();
                     for (Map.Entry<String, TimelyBalancedHost> e : metricToHostMap.entrySet()) {
+                        if (!nonCachedMetrics.contains(e.getKey())) {
                             writer.write(e.getKey());
                             writer.write(e.getValue().getHost());
                             writer.write(Integer.toString(e.getValue().getTcpPort()));
@@ -852,6 +991,7 @@ public class BalancedMetricResolver implements MetricResolver {
                             LOG.trace("Saving assigment: {} to {}:{}", e.getKey(), e.getValue().getHost(),
                                     e.getValue().getTcpPort());
                         }
+                    }
 
                     long now = System.currentTimeMillis();
                     assignmentsLastUpdatedLocal.set(now);
@@ -863,6 +1003,7 @@ public class BalancedMetricResolver implements MetricResolver {
                             metricToHostMap.size(), new Date(assignmentsLastUpdatedLocal.get()));
                 }
             } finally {
+                nonCachedMetricsLocalLock.readLock().unlock();
                 balancerLock.readLock().unlock();
             }
         } catch (Exception e) {
@@ -872,7 +1013,7 @@ public class BalancedMetricResolver implements MetricResolver {
                 writer.close();
             }
             try {
-                assignmentsLock.writeLock().release();
+                assignmentsIPRWLock.writeLock().release();
             } catch (Exception e) {
                 LOG.error(e.getMessage(), e);
             }

server/bin/timely-server.sh

@@ -8,11 +8,6 @@ else
         TCNATIVE_SUFFIX="so"
 fi
 
-# netty tcnative file reference
-#netty-tcnative-1.1.33.Fork18-linux-x86_64-fedora.jar -> libnetty-tcnative.so
-#netty-tcnative-1.1.33.Fork18-linux-x86_64.jar -> libnetty-tcnative.so
-#netty-tcnative-1.1.33.Fork18-osx-x86_64.jar -> libnetty-tcnative.jnilib
-
 THIS_DIR="${THIS_SCRIPT%/*}"
 NATIVE_DIR="${THIS_DIR}/META-INF/native"
 BASE_DIR=${THIS_DIR}/..
@@ -32,14 +27,15 @@ fi
 mkdir -p ${NATIVE_DIR}
 
 pushd ${BASE_DIR}/bin
-$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative*.jar META-INF/native/libnetty_tcnative.${NATIVE_SUFFIX}
+$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative*.jar META-INF/native/libnetty_tcnative.${TCNATIVE_SUFFIX}
-$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-transport-native-epoll*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${NATIVE_SUFFIX}
+$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-transport-native-epoll*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
 popd
 
 export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*"
 JVM_ARGS="-Xmx256m -Xms256m -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS}"
 JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
 JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
+JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}/libnetty-${TCNATIVE_SUFFIX}"
 
 echo "$JAVA_HOME/bin/java ${JVM_ARGS} timely.Server --spring.config.name=timely"
 $JAVA_HOME/bin/java ${JVM_ARGS} timely.Server --spring.config.name=timely

server/conf/log4j2-spring.xml

@@ -26,12 +26,12 @@
     </Console>
   </Appenders>
   <Loggers>
-    <Logger name="timely.subscription" level="trace" additivity="false">
+    <Root level="debug" includeLocation="false">
-      <AppenderRef ref="Subscription" />
-    </Logger>
-    <Root level="trace" includeLocation="false">
       <AppenderRef ref="Console" level="warn" />
-      <AppenderRef ref="Rolling" level="trace" />
+      <AppenderRef ref="Rolling" level="debug" />
     </Root>
+    <Logger name="timely" additivity="false">
+      <AppenderRef ref="Console" level="debug" />
+    </Logger>
   </Loggers>
 </Configuration>

server/conf/timely.yml

@@ -4,14 +4,14 @@ timely:
   metric-age-off-days:
     default: 7
   cache:
-    enabled:
+    enabled: true
     age-off-hours:
-      default: 30
+      default: 26
   metrics-report-ignored-tags:
   accumulo:
-    instance-name:
+    instance-name: accumulo
-    zookeepers:
+    zookeepers: localhost:2181
-    username:
+    username: root
     password:
     write:
       latency: 5s
@@ -22,23 +22,23 @@ timely:
       max-downsample-memory: 10000000
   security:
     session-max-age: 86400
-    allow-anonymous-access: false
+    allow-anonymous-access: true
-    ssl:
+    serverSsl:
       certificate-file:
       key-file:
       key-password:
-      use-generated-keypair: false
+      use-generated-keypair: true
       trust-store-file:
       use-openssl: true
       #use-ciphers:
   server:
-    ip:
+    ip: 0.0.0.0
-    tcp-port:
+    tcp-port: 4242
-    udp-port:
+    udp-port: 4542
   http:
-    host:
+    host: localhost
-    ip:
+    ip: 0.0.0.0
-    port:
+    port: 4243
     redirect-path: /secure-me
     strict-transport-max-age: 608400
     cors:
@@ -49,8 +49,8 @@ timely:
       allowed-methods: DELETE,GET,HEAD,OPTIONS,PUT,POST
       allowed-headers: content-type
   websocket:
-    ip:
+    ip: 0.0.0.0
-    port:
+    port: 4244
     timeout: 60
     subscription-lag: 120
     scanner-batch-size: 10000

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

@@ -1,5 +1,8 @@
 package timely;
 
+import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS;
+import static timely.store.cache.DataStoreCache.NON_CACHED_METRICS_LOCK_PATH;
+
 import java.io.File;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
@@ -63,6 +66,7 @@ import org.apache.curator.x.discovery.ServiceDiscovery;
 import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
 import org.apache.curator.x.discovery.ServiceInstance;
 import org.apache.curator.x.discovery.ServiceInstanceBuilder;
+import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.data.Stat;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -125,6 +129,7 @@ public class Server {
     private static final String OS_VERSION = "os.version";
     private static final String WS_PATH = "/websocket";
 
+    private CuratorFramework curatorFramework;
     protected static final CountDownLatch LATCH = new CountDownLatch(1);
     static ConfigurableApplicationContext applicationContext;
 
@@ -146,6 +151,9 @@ public class Server {
     protected volatile boolean shutdown = false;
     private final int DEFAULT_EVENT_LOOP_THREADS;
 
+    private String[] zkPaths = new String[] { SERVICE_DISCOVERY_PATH, NON_CACHED_METRICS,
+            NON_CACHED_METRICS_LOCK_PATH };
+
     private static boolean useEpoll() {
 
         // Should we just return true if this is Linux and if we get an error
@@ -175,13 +183,8 @@ public class Server {
         }
     }
 
-    public void registerService() {
+    public void registerService(CuratorFramework curatorFramework) {
         try {
-            RetryPolicy retryPolicy = new RetryForever(1000);
-            CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(config.getAccumulo().getZookeepers(),
-                    30000, 10000, retryPolicy);
-            curatorFramework.start();
-
             try {
                 Stat stat = curatorFramework.checkExists().forPath(SERVICE_DISCOVERY_PATH);
                 if (stat == null) {
@@ -369,6 +372,10 @@ public class Server {
             LOG.error("Error closing WebSocketRequestDecoder during shutdown", e);
         }
 
+        if (curatorFramework != null) {
+            curatorFramework.close();
+        }
+
         if (applicationContext != null) {
             LOG.info("Closing applicationContext");
             applicationContext.close();
@@ -390,13 +397,33 @@ public class Server {
         this.config = conf;
     }
 
+    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 {
 
+        RetryPolicy retryPolicy = new RetryForever(1000);
+        curatorFramework = CuratorFrameworkFactory.newClient(config.getAccumulo().getZookeepers(), 30000, 10000,
+                retryPolicy);
+        curatorFramework.start();
+        ensureZkPaths(curatorFramework, zkPaths);
+
         int nettyThreads = Math.max(1,
                 SystemPropertyUtil.getInt("io.netty.eventLoopThreads", Runtime.getRuntime().availableProcessors() * 2));
         dataStore = DataStoreFactory.create(config, nettyThreads);
         if (config.getCache().isEnabled()) {
-            dataStoreCache = new DataStoreCache(config);
+            dataStoreCache = new DataStoreCache(curatorFramework, config);
             dataStoreCache.setInternalMetrics(dataStore.getInternalMetrics());
             dataStore.setCache(dataStoreCache);
         }
@@ -500,7 +527,7 @@ public class Server {
             udpServer.option(EpollChannelOption.SO_REUSEPORT, true);
             udpChannelHandleList.add(udpServer.bind(udpIp, udpPort).sync().channel());
         }
-        registerService();
+        registerService(curatorFramework);
         shutdownHook();
         LOG.info(
                 "Server started. Listening on {}:{} for TCP traffic, {}:{} for HTTP traffic, {}:{} for WebSocket traffic, and {}:{} for UDP traffic",

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

@@ -7,11 +7,14 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.Timer;
 import java.util.TimerTask;
+import java.util.TreeSet;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.StampedLock;
 
 import io.netty.handler.codec.http.HttpResponseStatus;
@@ -21,7 +24,16 @@ import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 import org.apache.accumulo.core.security.Authorizations;
+import org.apache.commons.lang3.SerializationUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.recipes.atomic.DistributedAtomicValue;
+import org.apache.curator.framework.recipes.cache.TreeCache;
+import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
+import org.apache.curator.framework.recipes.cache.TreeCacheListener;
+import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
+import org.apache.curator.retry.RetryForever;
+import org.apache.zookeeper.CreateMode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import timely.api.request.AuthenticatedRequest;
@@ -45,12 +57,17 @@ import timely.store.iterators.RateIterator;
 public class DataStoreCache {
 
     private static final Logger LOG = LoggerFactory.getLogger(DataStoreCache.class);
+    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";
+
     public static final String DEFAULT_AGEOFF_KEY = "default";
 
     private Map<String, Map<TaggedMetric, GorillaStore>> gorillaMap = new HashMap<>();
     private StampedLock gorillaMapLock = new StampedLock();
 
     private Set<String> nonCachedMetrics = Collections.synchronizedSet(new HashSet<>());
+    private DistributedAtomicValue nonCachedMetricsIP = null;
+    private InterProcessReadWriteLock nonCachedMetricsIPRWLock;
     private long maxUniqueTagSets;
     private boolean anonAccessAllowed;
     private Map<String, Long> minimumAgeOff;
@@ -58,11 +75,16 @@ public class DataStoreCache {
     private int flushBatch = 0;
     private int numBatches = 5;
     private InternalMetrics internalMetrics;
-
     private Timer maintenanceTimer = new Timer("DataStoreCacheTimer");
 
-    public DataStoreCache(Configuration conf) {
+    public DataStoreCache(CuratorFramework curatorFramework, Configuration conf) {
-        nonCachedMetrics.addAll(conf.getCache().getNonCachedMetrics());
+        if (curatorFramework != null) {
+            addNonCachedMetricsListener(curatorFramework);
+        }
+        LOG.info("Adding initial values from configuration");
+        addNonCachedMetrics(conf.getCache().getNonCachedMetrics());
+        LOG.info("Reading initial values from nonCachedMetricsIP");
+        readNonCachedMetricsIP();
         maxUniqueTagSets = conf.getCache().getMaxUniqueTagSets();
         anonAccessAllowed = conf.getSecurity().isAllowAnonymousAccess();
         Map<String, Integer> cacheAgeOff = conf.getCache().getMetricAgeOffHours();
@@ -142,6 +164,124 @@ public class DataStoreCache {
         }, (60 * 1000), (60 * 1000));
     }
 
+    private void addNonCachedMetrics(Collection<String> nonCachedMetricsUpdate) {
+        if (!nonCachedMetricsUpdate.isEmpty()) {
+            try {
+                LOG.info("Adding {} to local nonCachedMetrics", nonCachedMetricsUpdate);
+                nonCachedMetrics.addAll(nonCachedMetricsUpdate);
+                if (nonCachedMetricsIP != null) {
+                    try {
+                        nonCachedMetricsIPRWLock.writeLock().acquire();
+                        byte[] currentNonCachedMetricsDistributedBytes = nonCachedMetricsIP.get().postValue();
+                        Set<String> currentNonCachedMetricsIP;
+                        if (currentNonCachedMetricsDistributedBytes == null) {
+                            currentNonCachedMetricsIP = new TreeSet<>();
+                        } else {
+                            try {
+                                currentNonCachedMetricsIP = SerializationUtils
+                                        .deserialize(currentNonCachedMetricsDistributedBytes);
+                            } catch (Exception e) {
+                                LOG.error(e.getMessage());
+                                currentNonCachedMetricsIP = new TreeSet<>();
+                            }
+                        }
+                        if (currentNonCachedMetricsIP.containsAll(nonCachedMetricsUpdate)) {
+                            LOG.info("nonCachedMetricsIP already contains {}", nonCachedMetricsUpdate);
+                        } else {
+                            nonCachedMetricsUpdate.removeAll(currentNonCachedMetricsIP);
+                            LOG.info("Adding {} to nonCachedMetricsIP", nonCachedMetricsUpdate);
+                            TreeSet<String> updateSet = new TreeSet<>();
+                            updateSet.addAll(currentNonCachedMetricsIP);
+                            updateSet.addAll(nonCachedMetricsUpdate);
+                            byte[] updateValue = SerializationUtils.serialize(updateSet);
+                            nonCachedMetricsIP.trySet(updateValue);
+                            if (!nonCachedMetricsIP.get().succeeded()) {
+                                nonCachedMetricsIP.forceSet(updateValue);
+                            }
+                        }
+                    } finally {
+                        nonCachedMetricsIPRWLock.writeLock().release();
+                    }
+                }
+            } catch (Exception e) {
+                LOG.error(e.getMessage(), e);
+            }
+        }
+    }
+
+    private void addNonCachedMetricsListener(CuratorFramework curatorFramework) {
+        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 = new TreeCacheListener() {
+
+            @Override
+            public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent event) throws Exception {
+                if (event.getType().equals(TreeCacheEvent.Type.NODE_UPDATED)) {
+                    LOG.info("Handling nonCachedMetricsIP event {}", event.getType().toString());
+                    readNonCachedMetricsIP();
+                }
+            }
+        };
+
+        try {
+            TreeCache nonCachedMetricsTreeCache = new TreeCache(curatorFramework, NON_CACHED_METRICS);
+            nonCachedMetricsTreeCache.getListenable().addListener(nonCachedMetricsListener);
+            nonCachedMetricsTreeCache.start();
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+        }
+    }
+
+    private void readNonCachedMetricsIP() {
+        try {
+            nonCachedMetricsIPRWLock.readLock().acquire();
+            byte[] nonCachedMetricsUpdateBytes = nonCachedMetricsIP.get().postValue();
+            Set<String> nonCachedMetricsUpdate;
+            if (nonCachedMetricsUpdateBytes == null) {
+                nonCachedMetricsUpdate = new TreeSet<>();
+            } else {
+                nonCachedMetricsUpdate = SerializationUtils.deserialize(nonCachedMetricsIP.get().postValue());
+            }
+
+            if (nonCachedMetrics.containsAll(nonCachedMetricsUpdate)) {
+                LOG.info("local nonCachedMetrics already contains {}", nonCachedMetricsUpdate);
+            } else {
+                nonCachedMetricsUpdate.removeAll(nonCachedMetrics);
+                LOG.info("Adding {} to local nonCachedMetrics", nonCachedMetricsUpdate);
+                nonCachedMetrics.addAll(nonCachedMetricsUpdate);
+            }
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+        } finally {
+            try {
+                nonCachedMetricsIPRWLock.readLock().release();
+            } catch (Exception e) {
+                LOG.error(e.getMessage());
+            }
+        }
+    }
+
+    private void testIPRWLock(CuratorFramework curatorFramework, InterProcessReadWriteLock lock, String path) {
+        try {
+            lock.writeLock().acquire(10, TimeUnit.SECONDS);
+        } catch (Exception e1) {
+            try {
+                curatorFramework.delete().deletingChildrenIfNeeded().forPath(path);
+                curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT)
+                        .forPath(path);
+            } catch (Exception e2) {
+                LOG.info(e2.getMessage(), e2);
+            }
+        } finally {
+            try {
+                lock.writeLock().release();
+            } catch (Exception e3) {
+                LOG.error(e3.getMessage());
+            }
+        }
+    }
+
     public void setInternalMetrics(InternalMetrics internalMetrics) {
         this.internalMetrics = internalMetrics;
     }
@@ -204,7 +344,7 @@ public class DataStoreCache {
                     LOG.info("Cache of metric {} has {} tag variations.  Discontinuing cache.", entry1.getKey(),
                             numberTagVariations);
                     metricItr.remove();
-                    nonCachedMetrics.add(entry1.getKey());
+                    addNonCachedMetrics(Collections.singleton(entry1.getKey()));
                 }
             }
         } finally {
@@ -396,6 +536,10 @@ public class DataStoreCache {
 
     private boolean shouldCache(Metric metric) {
         String metricName = metric.getName();
+        if (nonCachedMetrics.contains(metricName)) {
+            return false;
+        }
+
         long stamp = gorillaMapLock.readLock();
         try {
             if (gorillaMap.containsKey(metricName)) {
@@ -405,19 +549,17 @@ public class DataStoreCache {
             gorillaMapLock.unlockRead(stamp);
         }
 
-        if (nonCachedMetrics.contains(metricName)) {
+        Set<String> tempNonCachedMetrics = new LinkedHashSet<>();
+        synchronized (nonCachedMetrics) {
+            tempNonCachedMetrics.addAll(nonCachedMetrics);
+        }
+        for (String r : tempNonCachedMetrics) {
+            if (metricName.matches(r)) {
+                addNonCachedMetrics(Collections.singleton(metricName));
                 return false;
             }
+        }
 
-        synchronized (nonCachedMetrics) {
-            for (String r : nonCachedMetrics) {
-                if (metricName.matches(r)) {
-                    LOG.info("Adding {} to list of non-cached metrics", metricName);
-                    nonCachedMetrics.add(metricName);
-                    return false;
-                }
-            }
-        }
         return true;
     }
 

server/src/test/java/timely/store/cache/TestDataStoreCacheIterator.java

@@ -43,7 +43,7 @@ public class TestDataStoreCacheIterator {
 
     private DataStoreCache getMetricMemoryStore1(long baseTime) throws TimelyException {
 
-        DataStoreCache mmStore = new DataStoreCache(configuration);
+        DataStoreCache mmStore = new DataStoreCache(null, configuration);
 
         Map<String, String> tags = new HashMap<>();
         tags.put("part", "webservice");
@@ -69,7 +69,7 @@ public class TestDataStoreCacheIterator {
 
     private DataStoreCache getMetricMemoryStore2(long baseTime) throws TimelyException {
 
-        DataStoreCache mmStore = new DataStoreCache(configuration);
+        DataStoreCache mmStore = new DataStoreCache(null, configuration);
 
         int increment = 10;
         Map<String, String> tags = new HashMap<>();

server/src/test/java/timely/store/cache/TestMemoryDataStore.java

@@ -27,7 +27,7 @@ public class TestMemoryDataStore {
 
     private DataStoreCache getMetricMemoryStore1(long baseTimestamp) throws TimelyException {
 
-        DataStoreCache mmStore = new DataStoreCache(configuration);
+        DataStoreCache mmStore = new DataStoreCache(null, configuration);
 
         long timestamp = baseTimestamp + 10000;
         Map<String, String> tags = new HashMap<>();
@@ -61,7 +61,7 @@ public class TestMemoryDataStore {
 
     private DataStoreCache getMetricMemoryStore2(long baseTimestamp) throws TimelyException {
 
-        DataStoreCache mmStore = new DataStoreCache(configuration);
+        DataStoreCache mmStore = new DataStoreCache(null, configuration);
 
         Map<String, String> tags = new HashMap<>();
         tags.put("part", "webservice");
@@ -154,7 +154,7 @@ public class TestMemoryDataStore {
 
     @Test
     public void TestExtentOfStorage() {
-        DataStoreCache mmStore = new DataStoreCache(configuration);
+        DataStoreCache mmStore = new DataStoreCache(null, configuration);
 
         HashMap<String, String> tags = new HashMap<>();
         tags.put("host", "localhost");