                           ]
                          )))
               (write-locked (:endpoint-socket-lock worker)
-                (reset! (:task->node+port worker) my-assignment))
+                (reset! (:cached-task->node+port worker) my-assignment))
               (doseq [endpoint remove-connections]
-                (.close (@(:node+port->socket worker) endpoint)))
+                (.close (@(:cached-node+port->socket worker) endpoint)))
               (apply swap!
-                     (:node+port->socket worker)
+                     (:cached-node+port->socket worker)
                      dissoc
                      remove-connections)
           )))))

      :component->stream->fields (component->stream->fields (:system-topology <>))
      :component->sorted-tasks (->> (:task->component <>) reverse-map (map-val sort))
      :endpoint-socket-lock (mk-rw-lock)
      :node+port->socket (atom {})
      :task->node+port (atom {})
      :transfer-queue transfer-queue
      :executor-receive-queue-map executor-receive-queue-map
      :short-executor-receive-queue-map (map-key first executor-receive-queue-map)

(defn mk-transfer-tuples-handler [worker]
  (let [^DisruptorQueue transfer-queue (:transfer-queue worker)
        drainer (ArrayList.)
        node+port->socket (:cached-node+port->socket worker)
        task->node+port (:cached-task->node+port worker)
        endpoint-socket-lock (:endpoint-socket-lock worker)
        ]
    (fn [packets _ batch-end?]
