 
   (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]))))
+
+  (def res-multi-fail (run-job (update-in catalog-http-q&map&idx [0] assoc :onyx/restart-pred-fn ::restartable?) lifecycles-fail))
 
   (u/delete-indexes (.toString id)))
 

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

(deftest fault-logic
  (testing "Successfully processed all messages"
    (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)))))
(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}
