                                 (async/>!! in-chan data)))))
                       (and
                         (not (nil? in-data))
+                        (not (nil? pipeline))
                         (> batch-delay 0)
                         (> batch-size 0)) (create-batched-in-fn (get-in-fn pipeline))
-                      (not (nil? in-data)) (create-flood-in-fn (get-in-fn pipeline))
-                      :default (fn [] (.inc out-cntr))))
+                      (and
+                        (not (nil? in-data))
+                        (not (nil? pipeline))) (create-flood-in-fn (get-in-fn pipeline))
+                      :default (if (not (.contains scenario "async-pipeline"))
+                                 (fn [] (.inc out-cntr)))))
           async-out-count-loop (if
                                  (and
                                    (.endsWith scenario "-async-pipeline")

                                 (ProcessingLoop.
                                   "AsyncOutputCountLoop"
                                   (fn []
                                     (while true
                                       (do
                                         (async/<!! out-chan)
                                         (.inc out-cntr))))))
          async-pipeline (if
                           (and
                             (.endsWith scenario "-async-pipeline")
                             (not (nil? in-data)))
                           (async/pipeline
                             8

                             (update-stats self-adaptivity-controller counts))))))]
      (println "Starting experiment...")
      (.setName (Thread/currentThread) "Main")
      (run-repeat
        (executor)
        (fn []
