                  (or (not (empty? @pending-messages))
                      (not (empty? batch))))
         (if (:kafka/done-unsupported? event)
-          (throw (UnsupportedOperationException. ":done is not supported for auto assigned kafka partitions. 
-                                                  (:kafka/partition must be supplied)"))
+          (throw (ex-info ":done is not supported for auto assigned kafka partitions. (:kafka/partition must be supplied)"
+                          {:recoverable? false
+                           :task-map task-map}))
           (reset! drained? true)))
       {:onyx.core/batch batch}))
 

                        :n-peers n-peers
                        :min-peers min-peers
                        :max-peers max-peers
                        :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" {:recoverable? false
                                                               :policy policy}))))
    (checkpoint-or-beginning log consumer topic kpartition group-id task-map)))

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