     (is (= "bar" (-> res-native-idx first :_source :foo)))))
 
 (deftest fault-logic
-  (testing "Successfully processed all messages"
+  (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)))))+    (is (= -1 (:chunk-index task-chunk-restart))))
+  (testing "Successfully processed all messages with failure"
+    (is (= 11 (count res-multi-fail)))))
(def out-calls
  {:lifecycle/before-task-start inject-out-ch})

(def lifecycles
  [{:lifecycle/task :read-messages
    :lifecycle/calls :onyx.plugin.elasticsearch/read-messages-calls}

   {: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]
   (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]
   (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"})
