          {:lifecycle/task :identity
           :lifecycle/calls ::identity-crash}
          {:lifecycle/task :out
-          :lifecycle/calls :onyx.plugin.core-async/writer-calls}]]
-    (with-test-env [test-env [6 env-config peer-config]]
+          :lifecycle/calls :onyx.plugin.core-async/writer-calls}]
+
+
+        stats-holder (let [stats (map->MonitoringStats {})]
+                       (reduce (fn [st k] (assoc st k (atom []))) 
+                               stats
+                               (keys stats))) 
+
+        event-fn (fn print-monitoring-event [_ event]
+                   (let [stats-value (swap! (get stats-holder (:event event)) conj event)]
+                     #_(timbre/info "MONITORING NOW: " stats-value)))
+
+        monitoring-config {:monitoring :custom
+                           :zookeeper-read-catalog event-fn
+                           :window-log-playback event-fn
+                           :window-log-write-entry event-fn}]
+    (with-test-env [test-env [6 env-config peer-config monitoring-config]]
       (onyx.api/submit-job peer-config
                            {:catalog catalog
                             :workflow workflow

;;; delay is rather long

(def input
  [{:id 1  :age 21 :event-time #inst "2015-09-13T03:00:00.829-00:00"}
   {:id 2  :age 12 :event-time #inst "2015-09-13T03:04:00.829-00:00"}
   {:id 3  :age 3  :event-time #inst "2015-09-13T03:05:00.829-00:00"}
   {:id 4  :age 64 :event-time #inst "2015-09-13T03:06:00.829-00:00"}
   {:id 5  :age 53 :event-time #inst "2015-09-13T03:07:00.829-00:00"}
   {:id 6  :age 52 :event-time #inst "2015-09-13T03:08:00.829-00:00"}
   {:id 7  :age 24 :event-time #inst "2015-09-13T03:09:00.829-00:00"}
   {:id 8  :age 35 :event-time #inst "2015-09-13T03:15:00.829-00:00"}
   {:id 9  :age 49 :event-time #inst "2015-09-13T03:25:00.829-00:00"}
   {:id 10 :age 37 :event-time #inst "2015-09-13T03:45:00.829-00:00"}
   {:id 11 :age 15 :event-time #inst "2015-09-13T03:03:00.829-00:00"}
   {:id 12 :age 22 :event-time #inst "2015-09-13T03:56:00.829-00:00"}
   {:id 13 :age 83 :event-time #inst "2015-09-13T03:59:00.829-00:00"}
   {:id 14 :age 60 :event-time #inst "2015-09-13T03:32:00.829-00:00"}
   {:id 15 :age 35 :event-time #inst "2015-09-13T03:16:00.829-00:00"}])

(def expected-windows
  [[1442113200000 1442113499999 1] [1442113200000 1442113499999 2] [1442113500000 1442113799999 1] [1442113500000 1442113799999 2] [1442113500000 1442113799999 3] [1442114100000 1442114399999 1] [1442115000000 1442115299999 1] [1442116500000 1442116799999 1] [1442116500000 1442116799999 2] [1442113200000 1442113499999 3] [1442115900000 1442116199999 1] [1442114700000 1442114999999 1] [1442114100000 1442114399999 2] [1442113500000 1442113799999 4] [1442113500000 1442113799999 5]])


(defn update-atom! [event window-id lower-bound upper-bound state]
  (swap! test-state conj [lower-bound upper-bound state]))

(defn restartable? [e] 
  true)
