 (def out-calls
   {:lifecycle/before-task-start inject-out-ch})
 
+(def batch-num (atom 0))
+
+(def read-crash
+  {:lifecycle/before-batch
+   (fn [event lifecycle]
+     ; give the peer a bit of time to write the chunks out and ack the batches,
+     ; since we want to ensure that the batches aren't re-read on restart for ease of testing
+     (Thread/sleep 7000)
+     (when (= (swap! batch-num inc) 2)
+       (throw (ex-info "Restartable" {:restartable? true}))))})
+
 (def lifecycles
   [{:lifecycle/task :read-messages
     :lifecycle/calls :onyx.plugin.elasticsearch/read-messages-calls}

(def batch-size 20)

(def workflow [[:read-messages :out]])

;; Create catalogs with different search scenarios


    (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)))))