   p/BarrierSynchronization
   (synced? [this epoch]
     (when @exception (throw @exception))
-    (empty? (clear-write-futures! write-futures)))
+    (empty? (vswap! write-futures clear-write-futures!)))
 
   (completed? [this]
     (when @exception (throw @exception))
-    (empty? (clear-write-futures! write-futures)))
+    (empty? (vswap! write-futures clear-write-futures!)))
 
   p/Checkpointed
   (recover! [this _ _] 

(deftype ExceptionCallback [e]
  Callback
  (onCompletion [_ v exception]
    (reset! e exception)))

(defn write-messages [{:keys [onyx.core/task-map onyx.core/log-prefix] :as event}]
  (let [_ (s/validate onyx.tasks.kafka/KafkaOutputTaskMap task-map)

    true)
  (write-batch [this {:keys [onyx.core/results]} replica _]
    (when @exception (throw @exception))
    (vswap! write-futures
            (fn [fs]
              (-> fs
                  (clear-write-futures!)
                  (into (comp (mapcat :leaves)
                              (map (fn [msg]
                                     (send-async! producer 
                                                  (message->producer-record key-serializer-fn serializer-fn topic kpartition msg)
                                                  {:send-callback write-callback}))))
                        (:tree results)))))
    true))

(def write-defaults {:kafka/request-size 307200})
