                         :n-peers n-peers
                         :min-peers min-peers
                         :max-peers max-peers
+                        :recoverable? false
                         :task-map task-map})] 
         (log/error e)
         (throw e)))))

    (->KafkaWriteMessages task-map config topic producer serializer-fn)))

(defn read-handle-exception [event lifecycle lf-kw exception]
  :restart)

(def read-messages-calls
  {:lifecycle/before-task-start start-kafka-consumer

    (if (seq results)
      results
      (throw (ex-info "Could not locate any Kafka brokers to connect to."
                      {:recoverable? true
                       :zk-addr zk-addr})))))

(defn highest-offset-to-commit [offsets]
  (->> (sort offsets)
