   p/BarrierSynchronization
   (synced? [this _]
     true)
-  (completed? [this] (empty? segments))
-
+  (completed? [this]
+    (and (= (.get idx) (.get size))
+         (nil? full-pub)
+         (empty? queued-offers)))
   p/Output
-  (prepare-batch [this {:keys [onyx.core/results onyx.core/triggered task->group-by-fn] :as event} 
-                  replica messenger]
-    (.clear ^java.util.ArrayList segments)
+  (prepare-batch [this {:keys [onyx.core/results onyx.core/triggered] :as event} replica messenger]
+    (.set idx 0)
+    (.set size 0)
     (let [;; generate this on each new replica / messenger
           ; get-pub-fn (fn [segment dst-task-id]
           ;              (if-let [group-fn (task->group-by-fn dst-task-id)]

        pubs
        (recur (rest pubs))))))

(deftype MessengerOutput [^java.util.ArrayList segments 
                          ^:unsynchronized-mutable ^java.util.Iterator iterator
                          ^:unsynchronized-mutable queued-offers
                          ^:unsynchronized-mutable add-failed]
  p/Plugin
  (start [this event] this)
  (stop [this event] this)

          ;                      dest-pubs (m/task->publishers messenger dst-task-id)]
          ;                  (get dest-pubs (mod hsh (count dest-pubs))))
          ;                (rand-nth (m/task->publishers messenger dst-task-id))))
          ;; TODO, avoid initial flattening preprocessing step
          _ (run! (fn [result]
                    (run! (fn [seg]
                            (add-segment segments routes size seg event result))
                          (:leaves result)))
                  (:tree results))
          _ (run! (fn [seg] 
                    (add-segment segments routes size seg event {:leaves [seg]}))
                  triggered)]
      (run! pub/reset-segment-encoder! (m/publishers messenger))
      (set! queued-offers nil)
      (set! full-pub nil)
      true))

  (write-batch [this event replica messenger]
    (if (zero? (.get size))
      true
      (let [dst-task->publisher (m/task->publishers messenger)
            get-pub-fn (fn [dst-task-id] (get dst-task->publisher dst-task-id))]
        (cond full-pub
              ;; previous addition of segment to buffer resulted in an overflow
              ;; offer the full buffer, and then add the serialized segment to the reset buffer
              (if (neg? (pub/offer-segments! full-pub))
                false
                (do (pub/reset-segment-encoder! full-pub)
                    ;; publisher already has the last segment saved, lets encode it now
                    (when-not (pub/encode-segment! full-pub)
                      (throw (ex-info "Aeron buffer size is not big enough to contain the segment.
                                       Please increase the term buffer length via java property aeron.term.buffer.length"
                                      {})))
                    (set! full-pub nil)
                    (p/write-batch this event replica messenger)))

              (< (.get idx) (.get size))
              (loop [i (.get idx)] 
                (if (< i (.get size))
                  ;; add segment to corresponding buffer in publication
                  (let [segment (aget segments i)
                        route (aget routes i)
                        publisher ((get-pub-fn route) segment)]
                    (if (pub/encode-segment! publisher segment)
                      (recur (.incrementAndGet idx))
                      (do (set! full-pub publisher)
                          (p/write-batch this event replica messenger))))
                  (do
                   (set! queued-offers (m/publishers messenger))
                   (p/write-batch this event replica messenger))))

              :else
              (let [remaining (try-sends queued-offers)]
                (set! queued-offers remaining)
                (if (empty? remaining)
                  (do (run! (fn [i] 
                              (aset segments i nil)
                              (aset routes i nil))
                            (range (.get size)))
                      true)
                  false)))))))

(defn new-messenger-output [{:keys [onyx.core/task-map] :as event}]
  (let [bs (byte-array (max-message-length)) 
        segments (object-array 20000)
        routes (make-array clojure.lang.Keyword 20000)]
    (->MessengerOutput segments routes (AtomicInteger. 0) (AtomicInteger. 0) nil nil)))
