 (defn mk-transfer-tuples-handler [worker]
   (let [^DisruptorQueue transfer-queue (:transfer-queue worker)
         drainer (ArrayList.)
-        node+port->socket (:node+port->socket worker)
-        task->node+port (:task->node+port worker)
+        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?]

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

                    (doseq [[_ q] (:executor-receive-queue-map worker)]
                      (.haltProcessing q))
                    (doseq [executor executors] (.shutdown executor))
                    (doseq [[_ socket] @(:cached-node+port->socket worker)]
                      ;; this will do best effort flushing since the linger period
                      ;; was set on creation
                      (.close socket))
