             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 %})))))
+         #(extensions/emit (:onyx.core/monitoring event) {:event :peer/retry-message :latency %})))))
   event)
 
 (defn inject-batch-resources [compiled-before-batch-fn pipeline event]

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

                  (= ch retry-ch)
                  (->> (p-ext/retry-message pipeline event v)

                      results-by-acker))]
      (measure-latency
       #(extensions/internal-ack-messages messenger event link acks)
       #(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}]
