 (def catalog-native-idx
   (update-in catalog-http-idx [0] assoc :elasticsearch/client-type :native :elasticsearch/port es-native-port))
 
-(def out-chan (chan (sliding-buffer (inc n-messages))))
+(def out-chan (atom nil))
 
 (defn inject-out-ch [event lifecycle]
-  {:core.async/chan out-chan})
+  {:core.async/chan @out-chan})
 
 (def out-calls
   {:lifecycle/before-task-start inject-out-ch})

          (is (= -1 (:chunk-index task-chunk-restart))))))

    (with-test-env [test-env [2 env-config peer-config]]
      (let [job {:catalog catalog-http-q&map&idx
                 :workflow workflow
                 :lifecycles lifecycles-fail
                 :task-scheduler :onyx.task-scheduler/balanced}
            {:keys [job-id]} (onyx.api/submit-job peer-config job)]
        (onyx.api/await-job-completion peer-config job-id)
        (let [result (take-segments! out-chan 5000)]
          (is (= 11 (count result))))))))

  (esrd/create conn (.toString id) "_default_" {:foo "bar"})
  (Thread/sleep 5000)

  (testing "Successful Query for Native with Query and Index defined"
    (with-test-env [test-env [2 env-config peer-config]]
      (reset! out-chan (chan (sliding-buffer (inc n-messages))))
      (let [job {:catalog catalog-native-q&idx
                 :workflow workflow
                 :lifecycles lifecycles
                 :task-scheduler :onyx.task-scheduler/balanced}
            {:keys [job-id]} (onyx.api/submit-job peer-config job)]
        (onyx.api/await-job-completion peer-config job-id)
        (let [result (take-segments! @out-chan 5000)]
          (is (= "bar" (-> result first :_source :foo)))))))


  (testing "Successful Query for Native with Query, Map, and Index defined"
    (with-test-env [test-env [2 env-config peer-config]]
      (reset! out-chan (chan (sliding-buffer (inc n-messages))))
      (let [job {:catalog catalog-native-q&map&idx
                 :workflow workflow
                 :lifecycles lifecycles
                 :task-scheduler :onyx.task-scheduler/balanced}
            {:keys [job-id]} (onyx.api/submit-job peer-config job)]
        (onyx.api/await-job-completion peer-config job-id)
        (let [result (take-segments! @out-chan 5000)]
          (is (= "bar" (-> result first :_source :foo)))))))

  (testing "Successful Query for Native with Query defined"
    (with-test-env [test-env [2 env-config peer-config]]
      (reset! out-chan (chan (sliding-buffer (inc n-messages))))
      (let [job {:catalog catalog-native-q&all
                 :workflow workflow
                 :lifecycles lifecycles
                 :task-scheduler :onyx.task-scheduler/balanced}
            {:keys [job-id]} (onyx.api/submit-job peer-config job)]
        (onyx.api/await-job-completion peer-config job-id)
        (let [result (take-segments! @out-chan 5000)]
          (is (= "bar" (-> result first :_source :foo)))))))

  (testing "Successful Query for Native for all"
    (with-test-env [test-env [2 env-config peer-config]]
      (reset! out-chan (chan (sliding-buffer (inc n-messages))))
      (let [job {:catalog catalog-native-idx
                 :workflow workflow
                 :lifecycles lifecycles
                 :task-scheduler :onyx.task-scheduler/balanced}
            {:keys [job-id]} (onyx.api/submit-job peer-config job)]
        (onyx.api/await-job-completion peer-config job-id)
        (let [result (take-segments! @out-chan 5000)]
          (is (= "bar" (-> result first :_source :foo))))))))

(deftest fault-logic
