 (deftype ExceptionCallback [e]
   Callback
   (onCompletion [_ v exception]
-    (reset! e exception)))
+    (when 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]
              (doall 
               (into (remove realized? fs)
                     (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})

          :else
          (ProducerRecord. message-topic message-partition k (serializer-fn message)))))

(defn clear-write-futures! [fs]
  (doall (remove (fn [f] 
                   (assert (not (.isCancelled ^java.util.concurrent.Future f)))
                   (realized? f)) 
                 fs)))

(defrecord KafkaWriteMessages [task-map config topic kpartition producer key-serializer-fn serializer-fn write-futures exception write-callback]
  p/Plugin
