         (-> state 
             (set-context! {:barrier-opts {:completed? all-barriers-completed?}
                            :publishers (m/publishers messenger)})
-            (checkpoint-input!)
-            (checkpoint-state!)
             (advance)))
       (goto-next-batch! state))))
 

        (set-pipeline! next-pipeline)
        (advance))))

(defn recover-windows-state
  [state]
  (let [{:keys [recover] :as context} (get-context state)
        {:keys [log-prefix task-map windows triggers] :as event} (get-event state)

      (#{:input :function} task-type)         (conj {:lifecycle :offer-barriers
                                                     :fn offer-barriers
                                                     :blockable? true})
      (#{:input} task-type)                   (conj {:lifecycle :recover-input 
                                                     :fn recover-input})
      windowed?                               (conj {:lifecycle :recover-state 
                                                     :fn recover-state})
      (#{:input :function :output} task-type) (conj {:lifecycle :next-iteration
                                                     :fn next-iteration})
      (#{:input} task-type)                   (conj {:lifecycle :input-poll-barriers
                                                     :fn input-poll-barriers})
      (#{:input :function} task-type)         (conj {:lifecycle :prepare-offer-barriers
                                                     :fn prepare-offer-barriers})
      ;; TODO: double check that checkpoint doesn't occur immediately after recovery
      (#{:input} task-type)                   (conj {:lifecycle :checkpoint-input
                                                     :fn checkpoint-input
                                                     :blockable? true})
      windowed?                               (conj {:lifecycle :checkpoint-state
                                                     :fn checkpoint-state
                                                     :blockable? true})
      (#{:input :function} task-type)         (conj {:lifecycle :offer-barriers
                                                     :fn offer-barriers
                                                     :blockable? true})
