   (:import [org.apache.zookeeper KeeperException$BadVersionException]
            [java.util.concurrent.locks LockSupport]))
 
-(def coordinator-log (atom []))
-
 (defn input-publications [{:keys [peer-sites message-short-ids] :as replica} peer-id job-id]
   (let [allocations (get-in replica [:allocations job-id])
         input-tasks (get-in replica [:input-tasks job-id])

        next-write-version (if write-coordinate?
                             (write-coordinate write-version log tenancy-id job-id coordinates)
                             write-version)]
    (-> state
        (update :barrier    merge {:scheduled? false})
        (update :checkpoint merge {:initiated? false
                                   :write-version next-write-version}))))

(defn schedule-next-barrier [state barrier-period-ns]
  (swap! coordinator-log conj :schedule-next-barrier)
  (update state :barrier merge {:scheduled? true
                                :next-barrier-time (+ (System/nanoTime) barrier-period-ns)}))


           (>= (:min-epoch status) epoch))))

(defn completed-checkpoint [{:keys [checkpoint messenger job-id tenancy-id log] :as state}]
  (let [{:keys [epoch write-version]} checkpoint
        write-coordinate? (> epoch 0)
        coordinates {:tenancy-id tenancy-id
