 (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)

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

(defn clear-write-futures! [write-futures]
  (vswap! write-futures 
          (fn [fs] 
            (when-not (empty? (filter #(.isCancelled ^java.util.concurrent.Future %) fs))
              (throw (ex-info "Failed assertion: kafka writes cancelled. Rebooting." {})))
            (doall (remove realized? fs)))))

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

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

  (completed? [this]
    (when @exception (throw @exception))
    (empty? (vswap! write-futures clear-write-futures!)))

  p/Checkpointed
  (recover! [this _ _] 
