[aggregator] fix thread leak + dont always start scheduler (#129)

* [aggregator] fix thread leak + dont always start scheduler * [aggregator] remove commented lines * [circleci] tie cache to java version
DataDog · Dec 2, 2020 · bd4bbbb · bd4bbbb
1 parent 73818a8
commit bd4bbbb
Show file tree

Hide file tree

Showing 3 changed files with 23 additions and 6 deletions.
diff --git a/.circleci/config.yml b/.circleci/config.yml
@@ -1,22 +1,35 @@
 version: 2.1
 
+commands:
+  create_custom_cache_lock:
+    description: "Create custom cache lock for java version."
+    parameters:
+      filename:
+        type: string
+    steps:
+      - run:
+          name: Grab java version and dump to file
+          command: java -version > << parameters.filename >>
+
 default_steps: &default_steps
   steps:
     - checkout
+    - create_custom_cache_lock:
+        filename: java-version-lock.txt
 
     # Download and cache dependencies
     - restore_cache:
         keys:
-          - -v5-dependencies-{{ checksum "pom.xml" }}-{{ arch }}
+          - -v6-dependencies-{{ checksum "pom.xml" }}-{{ checksum "java-version-lock.txt" }}-{{ arch }}
           # fallback to using the latest cache if no exact match is found
-          - -v5-dependencies-
+          - -v6-dependencies-
 
     - run: mvn dependency:go-offline
 
     - save_cache:
         paths:
           - ~/.m2
-        key: -v5-dependencies-{{ checksum "pom.xml" }}-{{ arch }}
+        key: -v6-dependencies-{{ checksum "pom.xml" }}-{{ checksum "java-version-lock.txt" }}-{{ arch }}
     - run: |
         mvn clean install -Dgpg.skip $MVN_EXTRA_OPTS
 

diff --git a/src/main/java/com/timgroup/statsd/StatsDAggregator.java b/src/main/java/com/timgroup/statsd/StatsDAggregator.java
@@ -23,14 +23,13 @@ public class StatsDAggregator {
             Arrays.asList(Message.Type.COUNT, Message.Type.GAUGE, Message.Type.SET));
     protected final ArrayList<Map<Message, Message>> aggregateMetrics;
 
-    // protected final Map<Message, Message> aggregateMetrics = new HashMap<>();
-    protected final Timer scheduler = new Timer(AGGREGATOR_THREAD_NAME, true);
-
     protected final int shardGranularity;
     protected final long flushInterval;
 
     private final StatsDProcessor processor;
 
+    protected Timer scheduler = null;
+
     private Telemetry telemetry;
 
     private class FlushTask extends TimerTask {
@@ -54,6 +53,10 @@ public StatsDAggregator(final StatsDProcessor processor, final int shards, final
         this.shardGranularity = shards;
         this.aggregateMetrics = new ArrayList<>(shards);
 
+        if (flushInterval > 0) {
+            this.scheduler = new Timer(AGGREGATOR_THREAD_NAME, true);
+        }
+
         for (int i = 0 ; i < this.shardGranularity ; i++) {
             this.aggregateMetrics.add(i, new HashMap<Message, Message>());
         }

diff --git a/src/main/java/com/timgroup/statsd/StatsDProcessor.java b/src/main/java/com/timgroup/statsd/StatsDProcessor.java
@@ -182,6 +182,7 @@ boolean isShutdown() {
 
     void shutdown() {
         shutdown = true;
+        aggregator.stop();
         executor.shutdown();
     }
 }