                       results-by-acker))]
       (measure-latency
        #(extensions/internal-ack-messages messenger event link acks)
-       #(extensions/emit (:onyx.core/monitoring) {:event :peer/ack-messages :latency %}))))
+       #(extensions/emit (:onyx.core/monitoring event) {:event :peer/ack-messages :latency %}))))
   event)
 
 (defn flow-retry-messages [replica state messenger {:keys [onyx.core/results] :as event}]

            link (operation/peer-link @replica state event (:completion-id root))]
        (measure-latency
         #(extensions/internal-retry-message messenger event (:id root) link)
         #(extensions/emit (:onyx.core/monitoring) {:event :peer/retry-message :latency %})))))
  event)

(defn inject-batch-resources [compiled-before-batch-fn pipeline event]

  (when (sentinel-found? event)
    (if (p-ext/drained? pipeline event)
      (do (complete-job event)
          (extensions/emit (:onyx.core/monitoring event) {:event :peer/try-complete-job}))
      (p-ext/retry-message pipeline event (sentinel-id event))))
  event)

(defn strip-sentinel
  [event]
  (if (= (:onyx/type (:onyx.core/task-map event)) :input)
    (do (extensions/emit (:onyx.core/monitoring event) {:event :peer/strip-sentinel})
        (update-in event
                   [:onyx.core/batch]
                   (fn [batch]
