   (next-replica! [this new-replica]
     (if (= replica new-replica)
       this
-      (let [job-id (:onyx.core/job-id event)
+      (let [{:keys [onyx.core/job-id onyx.core/task-id onyx.core/job-id onyx.core/id onyx.core/task-kill-flag]} event
             old-version (get-in replica [:allocation-version job-id])
-            new-version (get-in new-replica [:allocation-version job-id])
-            next-coordinator (coordinator/next-state coordinator replica new-replica)]
-        (if (or (= old-version new-version)
-                ;; wait for re-allocation
-                (killed? this)
-                (not= job-id 
-                      (:job (common/peer->allocated-job (:allocations new-replica) 
-                                                        (:onyx.core/id event)))))
-          (-> this 
-              (set-coordinator! next-coordinator)
-              (set-replica! new-replica))
-          (let [next-messenger (ms/next-messenger-state! messenger event replica new-replica)]
-            (assert (m/subscriber next-messenger) [:old-replica replica
-                                                  :new-replica new-replica])
-            (checkpoint/cancel! (:onyx.core/storage event))
-            (-> this
-                (set-sealed! false)
-                (set-messenger! next-messenger)
-                (set-coordinator! next-coordinator)
-                (set-replica! new-replica)
-                (reset-event!)
-                (goto-recover!)))))))
+            new-version (get-in new-replica [:allocation-version job-id])]
+        (cond (= old-version new-version)
+              (-> this 
+                  (set-coordinator! (coordinator/next-state coordinator replica new-replica))
+                  (set-replica! new-replica))
+
+              (let [allocated (common/peer->allocated-job (:allocations new-replica) id)] 
+                (or (killed? this)
+                    (not= task-id (:task allocated))
+                    (not= job-id (:job allocated))))
+              ;; Manually hit the kill switch early since we've been reallocated
+              ;; and we want to escape ASAP
+              (do
+               (reset! task-kill-flag true)
+               this)
+
+              :else
+              (let [next-messenger (ms/next-messenger-state! messenger event replica new-replica)]
+                (assert (m/subscriber next-messenger) [:old-replica replica
+                                                       :new-replica new-replica])
+                (checkpoint/cancel! (:onyx.core/storage event))
+                (-> this
+                    (set-sealed! false)
+                    (set-messenger! next-messenger)
+                    (set-coordinator! (coordinator/next-state coordinator replica new-replica))
+                    (set-replica! new-replica)
+                    (reset-event!)
+                    (goto-recover!)))))))
   (set-windows-state! [this new-windows-state]
     (set! windows-state new-windows-state)
     this)

      (>!! outbox-ch entry)))
  state)

(defn time-out-peers! [state peers]
  (if-not (empty? peers)
    (let [{:keys [onyx.core/log-prefix onyx.core/id onyx.core/task] :as event} (get-event state)
          timeout-msg (format "%s Peers timed out %s" log-prefix (vec peers) id task)
          next-state (reduce evict-dead-peer! state peers)] 
      (-> timeout-msg (doto println) (doto info))
      ;; backoff for a while so we don't repeatedly write leave messages


(def DEBUG false)

(defn iteration [state n-iters]
  ;(when DEBUG (viz/update-monitoring! state-machine))
  (loop [state (exec state) n n-iters]
    ;(print-state state)
    ; (when (zero? (rand-int 10000)) 
    ;   (print-state state))
    (if (and (advanced? state) (pos? n))
      (recur (exec state) ;; we could unroll exec loop a bit
             (if (new-iteration? state) 
               (dec n)
               n))
      state)))

(def task-iterations 1)

(defn run-task-lifecycle!
  "The main task run loop, read batch, ack messages, etc."
