                  :task-id task-id
                  :slot-id slot-id
                  :ledger-id new-ledger-id}})
-    (while (and (first (alts!! [kill-ch task-kill-ch] :default true))
-                (not= new-ledger-id (last (event->ledger-ids event))))
-      (info "New ledger id has not been published yet. Backing off.")
-      (Thread/sleep (arg-or-default :onyx.bookkeeper/ledger-id-written-back-off peer-opts))))) 
+    (loop []
+      (let [exit? (nil? (first (alts!! [kill-ch task-kill-ch] :default true)))
+            not-added? (not= new-ledger-id (last (event->ledger-ids event)))]
+        (cond exit? 
+              (info "Exiting assign-bookkeeper-log-id-spin early as peer has been reassigned.")
+              not-added? 
+              (do
+                (info "New ledger id has not been published yet. Backing off.")
+                (Thread/sleep (arg-or-default :onyx.bookkeeper/ledger-id-written-back-off peer-opts))
+                (recur))))))) 
 
 (defmethod state-extensions/initialize-log :bookkeeper [log-type {:keys [onyx.core/peer-opts] :as event}] 
   (let [bk-client (bookkeeper peer-opts)

          (.asyncAddEntry compacted-ledger 
                          compacted-serialized
                          HandleWriteCallback
                          (fn []
                            (close-handle previous-handle)
                            (emit-latency-value :window-log-compaction monitoring (- (System/currentTimeMillis) start-time))
                            (>!! outbox-ch
                                 {:fn :compact-bookkeeper-log-ids
                                  :args {:job-id job-id
                                         :task-id task-id
                                         :slot-id slot-id
                                         :peer-id id
                                         :prev-ledger-ids (vec (butlast current-ids))
                                         :new-ledger-ids [compacted-ledger-id]}}))))))))

(defn ch->type [ch batch-ch timeout-ch kill-ch task-kill-ch]
  (cond (= ch timeout-ch)
        :timeout
        (or (= ch kill-ch) (= ch task-kill-ch))
        :shutdown
        :else
        :read))

(defn take-write-batch [peer-opts batch-ch kill-ch task-kill-ch]
  (let [batch-size (arg-or-default :onyx.bookkeeper/write-batch-size peer-opts)
        timeout-ms (arg-or-default :onyx.bookkeeper/write-batch-timeout peer-opts)
        timeout-ch (timeout timeout-ms)]
    (loop [entries [] ack-fns [] i 0]
      (if (< i batch-size)
        (let [[[entry ack-fn] ch] (alts!! [kill-ch task-kill-ch batch-ch timeout-ch] :priority true)]
          (if entry 
            (recur (conj entries entry) 
                   (conj ack-fns ack-fn) 
                   (inc i))
            (let [msg-type (ch->type ch batch-ch timeout-ch kill-ch task-kill-ch)]
              [msg-type entries ack-fns])))
        [:read entries ack-fns]))))

(defn process-batches [{:keys [ledger-handle next-ledger-handle batch-ch] :as log} 
                       {:keys [onyx.core/kill-ch onyx.core/task-kill-ch onyx.core/peer-opts] :as event}]
  (thread 
    (loop [[result batch ack-fns] (take-write-batch peer-opts batch-ch kill-ch task-kill-ch)]
      ;; Safe point to transition to the next ledger handle
      (when @next-ledger-handle
        (compaction-transition log event))
      (when-not (empty? batch) 
        (.asyncAddEntry ^LedgerHandle @ledger-handle 
                        ^bytes (nippy/window-log-compress batch)
                        HandleWriteCallback
                        (fn [] (run! (fn [f] (f)) ack-fns))))
      (if-not (= :shutdown result)
        (recur (take-write-batch peer-opts batch-ch kill-ch task-kill-ch))))
    (info "BookKeeper: shutting down batch processing")))

(defn assign-bookkeeper-log-id-spin [{:keys [onyx.core/peer-opts
                                             onyx.core/job-id onyx.core/task-id


(def HandleWriteCallback
  (reify AsyncCallback$AddCallback
    (addComplete [this rc lh entry-id [success-fn fail-fn]]
      (if (= rc (BKException$Code/OK))
        (success-fn)
        (do (warn "Unable to complete async write to bookkeeper. BookKeeper exception code:" rc)
            (fail-fn))))))

(defn compaction-transition 
  "Transitions to a new compacted ledger, plus a newly created ledger created
