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

            (cp/seek-to-end-offset! consumer [{:topic topic :partition kpartition}])

            :else
            (throw (ex-info "Tried to seek to unknown policy" {:policy policy}))))
    (checkpoint-or-beginning log consumer topic kpartition group-id task-map)))

(defn log-id [replica-val job-id peer-id task-id]

        k (some-> m :key serializer-fn)
        p (some-> m :partition int)
        message-topic (get m :topic topic)]
    (cond (nil? message)
          (throw (ex-info "Payload is missing required. Need message key :message"
                          {:payload m}))

          (nil? message-topic)
          (throw (ex-info
                  (str "Unable to write message payload to Kafka! "
                       "Both :kafka/topic, and :topic in message payload "
                       "are missing!")
                  {:payload m}))
          :else
          (ProducerRecord. message-topic p k (serializer-fn message)))))

(defrecord KafkaWriteMessages [task-map config topic producer serializer-fn]
  p-ext/Pipeline
