                    ;; added the watch.
                    (when (zk/exists conn path)
                      (>!! read-ch true))
-                   (<!! read-ch)
-                   (close! read-ch)
-                   ;; Requires one more check. Watch may have been triggered by a delete
-                   ;; from a GC call.
-                   (if (zk/exists conn path)
-                     (seek-and-put-entry! log position ch)
-                     (recur)))))
+                   (let [[_ active-ch] (alts!! [read-ch kill-ch])]
+                     (when (= active-ch read-ch)
+                       (close! read-ch)
+                       ;; Requires one more check. Watch may have been triggered by a delete
+                       ;; from a GC call.
+                       (if (zk/exists conn path)
+                         (seek-and-put-entry! log position ch)
+                         (recur)))))))
              (recur (inc position)))))
        (catch java.lang.IllegalStateException e
          (trace e)

      (seek-to-new-origin! log ch))))

(defmethod extensions/subscribe-to-log ZooKeeper
  [{:keys [conn opts prefix] :as log} ch]
  (let [rets (chan)]
    (thread
     (try

      (zk/create conn (exception-path onyx-id) :persistent? true)

      (initialize-origin! conn config onyx-id)
      (assoc component :server server :conn conn :prefix onyx-id :kill-ch kill-ch)))

  (stop [component]
    (taoensso.timbre/info "Stopping ZooKeeper" (if (:zookeeper/server? config) "server" "client connection"))
    (zk/close (:conn component))
    (close! (:kill-ch component))

    (when (:server component)
      (.close ^TestingServer (:server component)))
