   (onCompletion [_ v exception]
     (when exception (reset! e exception))))
 
-(defn write-messages [{:keys [onyx.core/task-map onyx.core/log-prefix] :as event}]
+(defn write-messages [{:keys [onyx.core/task-map onyx.core/log-prefix]}]
   (let [_ (s/validate onyx.tasks.kafka/KafkaOutputTaskMap task-map)
         brokers (or (:kafka/bootstrap-servers task-map) (find-brokers task-map))
         request-size (or (get task-map :kafka/request-size) (get write-defaults :kafka/request-size))

              part (.partition rec)]
          (.set watermark (max (.get watermark) (.timestamp rec)))

          (cond (= :done deserialized) ;; TODO: Remove this in favor of target-offsets
                (do (swap! drained assoc (.partition rec) :emitted)
                    nil)


          key-deserializer (h/byte-array-deserializer)
          value-deserializer (h/byte-array-deserializer)
          consumer* (h/build-consumer consumer-config key-deserializer value-deserializer)
          partitions (mapv :partition 
                           (or (keys (:kafka/target-offsets task-map))
                               (h/partitions-for-topic consumer* topic)))
          n-partitions (count partitions)]
      (check-num-peers-equals-partitions task-map n-partitions)
      (let [kpartitions* (assign-partitions-to-slot! consumer* task-map topic n-partitions (:onyx.core/slot-id event))]
