    {:lifecycle/task :out
     :lifecycle/calls :onyx.plugin.core-async/writer-calls}])
 
+(def lifecycles-fail
+  [{:lifecycle/task :read-messages
+    :lifecycle/calls :onyx.plugin.elasticsearch/read-messages-calls}
+   {:lifecycle/task :read-messages
+    :lifecycle/calls ::read-crash}
+   {:lifecycle/task :out
+    :lifecycle/calls ::out-calls}
+   {:lifecycle/task :out
+    :lifecycle/calls :onyx.plugin.core-async/writer-calls}])
+
 (def v-peers (onyx.api/start-peers 2 peer-group))
 
 (defn submit-and-wait
-  [catalog]
-  (let [job-info (onyx.api/submit-job
-                   peer-config
-                   {:catalog catalog
-                    :workflow workflow
-                    :lifecycles lifecycles
-                    :task-scheduler :onyx.task-scheduler/balanced})]
-    (onyx.api/await-job-completion peer-config (:job-id job-info))
-    job-info))
+  ([catalog]
+   (submit-and-wait catalog lifecycles))
+  ([catalog lc]
+   (let [job-info (onyx.api/submit-job
+                    peer-config
+                    {:catalog catalog
+                     :workflow workflow
+                     :lifecycles lc
+                     :task-scheduler :onyx.task-scheduler/balanced})]
+     (onyx.api/await-job-completion peer-config (:job-id job-info))
+     job-info)))
 
 (defn run-job
-  [catalog]
-  (submit-and-wait catalog)
-  (take-segments! out-chan 5000))
+  ([catalog]
+   (run-job catalog lifecycles))
+  ([catalog lc]
+   (submit-and-wait catalog lc)
+   (take-segments! out-chan 5000)))
 
 (let [conn (u/connect-rest-client)]
   (esrd/create conn (.toString id) "_default_" {:foo "bar"})


  (let [job-info-restart (submit-and-wait (update-in catalog-http-q&map&idx [0] assoc :elasticsearch/restart-on-fail true))]
    (def task-chunk-restart (extensions/read-chunk (:log env) :chunk (get-in job-info-restart [:task-ids :read-messages :id]))))

  (u/delete-indexes (.toString id)))


    (is (= "bar" (-> res-native-idx first :_source :foo)))))

(deftest fault-logic
  (testing "Successfully processed all messages no failure"
    (is (= 11 (count res-multi))))
  (testing "Successfully wrote status to log for restart-on-fail=false"
    (is (= :complete (:status task-chunk-offset))))
  (testing "Updates not written to log for restart-on-fail=true"
    (is (= -1 (:chunk-index task-chunk-restart))))
  (testing "Successfully processed all messages with failure"
    (is (= 11 (count res-multi-fail)))))