      (k-cluster/all-brokers zk-utils))))
 
 (defn find-brokers [zk-addr]
-  (let [results (vals (id->broker zk-addr))]
+  (let [zk-addr (:kafka/zookeeper task-map)
+        results (vals (id->broker zk-addr))]
     (if (seq results)
       results
-      (throw (ex-info "Could not locate any Kafka brokers to connect to."
-                      {:recoverable? true
-                       :zk-addr zk-addr})))))
+      (do
+       (info "Could not locate any Kafka brokers to connect to. Backing off.")
+       (Thread/sleep (or (:kafka/unable-to-find-broker-backoff-ms task-map) 
+                         (:kafka/unable-to-find-broker-backoff-ms defaults)))
+       (throw (ex-info "Could not locate any Kafka brokers to connect to."
+                       {:recoverable? true
+                        :zk-addr zk-addr}))))))
 
 (defn check-num-peers-equals-partitions 
   [{:keys [onyx/min-peers onyx/max-peers onyx/n-peers kafka/partition] :as task-map} n-partitions]

(def defaults
  {:kafka/receive-buffer-bytes 65536
   :kafka/commit-interval 2000
   :kafka/wrap-with-metadata? false})

(defn seek-offset! [log-prefix consumer kpartition task-map topic checkpoint]
  (let [policy (:kafka/offset-reset task-map)


  (next-state [this _]
    (if (and iter (.hasNext ^java.util.Iterator iter))
      (let [rec ^ConsumerRecord (.next ^java.util.Iterator iter)
            new-offset (if rec
                         (.offset rec)
                         offset)]
