   [chost eshost path tenant from to]
   (println "Start time:" (tl/format-local-time (tl/local-now) :rfc822))
   (newline)
-  (time (bench-itself chost eshost path tenant from to))
+  (try
+    (time (bench-itself chost eshost path tenant from to))
+    (catch Exception e
+      (shutdown-agents)
+      (throw e)))
   (newline)
   (println "Finish time:" (tl/format-local-time (tl/local-now) :rfc822)))
 

  [session fetch! paths tenant rollup period from to]
  (let [futures
        (doall (map #(future
                       (->> (alia/execute
                             session fetch!
                             {:values [% tenant (int rollup)
                                       (int period)
                                       from to]
                              :fetch-size Integer/MAX_VALUE})
                            (map detect-aggregate)
                            (doall)
                            (seq)))
                    paths))]
    (map deref futures)))

(defn par-fetch-pmap
  "Fetch data in parallel fashion using pmap."
  [session fetch! paths tenant rollup period from to]

  (pmap (fn [path] (->> (alia/execute
                         session fetch!
                         {:values [path tenant (int rollup)

(defn c-get-data
  "Get data from C*."
  [host paths tenant rollup period from to]
  (println "Connecting to Cassandra...")
  (let [session (-> (alia/cluster {:contact-points [host]})
                    (alia/connect keyspace))
        fetch! (fetchq session)]
    (println "Getting data form Cassandra...")
    (let [data (time (doall (par-fetch session fetch! paths tenant
                                       rollup period from to)))]
      (newline)
      data)))

;;------------------------------------------------------------------------------
;; ElasticSearch
