                         :onyx/fn :onyx-peers.functions.functions/add-job-num
                         :jepsen/job-num job-num
                         :onyx/params [:jepsen/job-num]
-                        :onyx/restart-pred-fn :onyx-peers.functions.functions/restartable?
                         :onyx/type :function
                         :onyx/batch-size batch-size}]
              :lifecycles [{:lifecycle/task :persist
+                           :lifecycle/calls ::restart-calls}
+                          {:lifecycle/task :identity-log
+                           :lifecycle/calls ::restart-calls}
+                          {:lifecycle/task :persist
                            :lifecycle/calls :onyx.plugin.bookkeeper/write-ledger-calls}]
              :workflow [[:identity-log :persist]]
              :task-scheduler :onyx.task-scheduler/balanced}] 

                         :bookkeeper/deserializer-fn :onyx.compression.nippy/zookeeper-decompress
                         :bookkeeper/password-bytes password 
                         :bookkeeper/no-recovery? true
                         :onyx/restart-pred-fn :onyx-peers.functions.functions/restartable?
                         :onyx/max-peers 1
                         :onyx/batch-size batch-size
                         :onyx/doc "Reads a sequence from a BookKeeper ledger"})

(ns onyx-jepsen.simple-job
  (:require [taoensso.timbre :refer [info error debug fatal]]))

;; TODO, tighten up to test whether onyx-bookkeeper handles all its exceptions properly
(defn handle-exception [event lifecycle lf-kw exception]
  :restart)

(def restart-calls
  {:lifecycle/handle-exception handle-exception})

(defn add-read-ledgers [job batch-size zk-addr ledgers-root-path password ledger-ids]
  (let [read-ledger-task-names (mapv (fn [id]
