 (defn get-written-ledgers [onyx-client job-data onyx-id]
   (onyx.plugin.bookkeeper/read-ledgers-data (:log onyx-client) 
                                             onyx-id 
-                                            (:job-id @job-data)
-                                            (get-in @job-data [:task-ids :persist :id])))
-
-(defrecord WriteLogClient [client job-data ledger-handles ledger-handle ledger-ids onyx-client]
+                                            (:job-id job-data)
+                                            (get-in job-data [:task-ids :persist :id])))
+
+
+(defn close-ledger-handles [ledger-handles]
+  (mapv (fn [h] 
+          (try (.close h)
+               (catch Throwable t
+                 (info "Error closing handle" t)))) 
+        ledger-handles))
+
+(defn await-jobs-completions [peer-config jobs-data]
+  (mapv (fn [job-data]
+          (onyx.api/await-job-completion peer-config (:job-id job-data)))
+        jobs-data)) 
+
+(defrecord WriteLogClient [client jobs-data ledger-handles ledger-handle ledger-ids onyx-client]
   client/Client
   (setup! [this test node]
     (let [client (bookkeeper-client)

    (let [zk-addr (:zookeeper/address env-config)
          onyx-id (:onyx/id env-config)] 
      (case (:f op)
        :read-peer-log (timeout 4000
                                (assoc op :type :info :value :timed-out)
                                (try
                                  (assoc op 

                             (try
                               (assoc op
                                      :type :ok
                                      :value (let [ledgers (nth (partition-all (/ (count @ledger-ids) 
                                                                                  (:n-jobs op)) 
                                                                               @ledger-ids)
                                                                (:job-num op))
                                                   job-data (->> (simple-job/build-job (:params op) 
                                                                                       zk-addr
                                                                                       (onyx.log.zookeeper/ledgers-path onyx-id)
                                                                                       ledgers)
                                                                 (onyx.api/submit-job peer-config))]
                                               (swap! jobs-data conj job-data)
                                               job-data))
                               (catch Throwable t
                                 (assoc op :type :info :value t)))) 

