      (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]

    [true this])

  (write-batch [this {:keys [onyx.core/results]} replica _]
    (let [messages (mapcat :leaves (:tree results))]
      (->> messages
           (map (fn [msg]
                  (->> (:message msg)
                       (message->producer-record serializer-fn topic)
                       (send-async! producer))))
           (doall)
           ;; could perform the deref in synchonized? to block less often
           (run! deref))

  p/Plugin
  (start [this event]
    (let [{:keys [kafka/group-id kafka/consumer-opts]} task-map
          brokers (find-brokers task-map)


          _ (s/validate onyx.tasks.kafka/KafkaInputTaskMap task-map)
          consumer-config (merge {:bootstrap.servers (find-brokers (:kafka/zookeeper task-map))
                                  :group.id group-id
