   (when-not (or (= 1 (:onyx/max-peers task-map))
                 (= 1 (:onyx/n-peers task-map)))
     (throw (ex-info "Read datoms tasks must set :onyx/max-peers 1" task-map)))
+
   (let [_ (extensions/write-chunk log :chunk {:chunk-index -1 :status :incomplete} task-id)
         content (extensions/read-chunk log :chunk task-id)]
     (if (= :complete (:status content))

            [onyx.static.default-vals :refer [defaults]]
            [onyx.static.uuid :refer [random-uuid]]
            [onyx.extensions :as extensions]
            [taoensso.timbre :refer [info debug fatal]]))

(defn safe-connect [task-map]
  (if-let [uri (:datomic/uri task-map)]


(defn write-datoms [pipeline-data]
  (let [task-map (:onyx.core/task-map pipeline-data)
        _ (s/validate DatomicWriteDatomsTaskMap task-map)
        conn (safe-connect task-map)
        partition (:datomic/partition task-map)]
    (->DatomicWriteDatoms conn partition)))
