                              (.toArray ^clojure.lang.ArraySeq (:args ctx))
                              (into-array []))]
             (letfn [(clos [stgs ctx]
-                      (if-let [^RealizedStage s (first stgs)]
-                        (let [[^CompletableFuture next-args-future new-ctx] (run-task2 s ctx error-handler)
+                      (if-let [^RealizedStage current-stage (first stgs)]
+                        (let [current-stage-name (:name current-stage)
+                              error-handler (stage-name-aware-error-handler current-stage-name user-error-handler)
                               processor (Thread/currentThread)]
-                          (.handle ^CompletableFuture next-args-future
+                          (.handle (run-stage current-stage ctx)
                                    (reify BiFunction
                                      (apply [_ next-args ex]
                                        (if (= processor (Thread/currentThread))
-                                         (process-result (assoc new-ctx :args next-args) stgs)
+                                         (-> (get-next-ctx ctx next-args ex error-handler)
+                                             (process-result stgs))
                                          (.submit ^ExecutorService executor
                                                   (wrap-with-arguments-aware-callable
                                                     (fn []
-                                                      (let [ctx (if (some? ex)
-                                                                  (process-exception new-ctx (:args new-ctx) ex (:name s) error-handler)
-                                                                  (assoc new-ctx :args next-args))]
-                                                        (process-result ctx stgs)))
+                                                      (-> (get-next-ctx ctx next-args ex error-handler)
+                                                          (process-result stgs)))
                                                     args-array)))))))
                         (do
                           (when post-hook (post-hook ctx))

                              timeout-val))))
  (stop! [this]))

(defn- run-task2 [^RealizedStage current-stage ctx error-handler]
  (let [task (.stage-fn current-stage)
        args (:args ctx)
        args (if (or (nil? args) ;; empty arguments

                        (if (or (:abort ctx) use-fork-or-join?)
                          (do
                            (when post-hook (post-hook ctx))
                            (when (and user-error-handler use-fork-or-join?)
                              (try
                                (user-error-handler (UnsupportedOperationException. "Fork or Join is not supported in AsyncDedicatedThreadPoolPipeline"))
                                (catch Exception _)))
                            (.complete result ctx))
                          (clos (rest stgs) ctx))))]
