               new-connections (set/difference needed-connections current-connections)
               remove-connections (set/difference current-connections needed-connections)]
               (swap! (:cached-node+port->socket worker)
-                     merge
+                     #(HashMap. (merge (into {} %1) %2))
                      (into {}
-                       (dofor [[node port :as endpoint] new-connections]
-                         [endpoint
+                       (dofor [endpoint-str new-connections
+                               :let [[node port] (string->endpoint endpoint-str)]]
+                         [endpoint-str
                           (msg/connect
                            (:mq-context worker)
                            storm-id

        (this (fn [& ignored] (schedule (:timer worker) 0 this))))
      ([callback]
        (let [assignment (.assignment-info storm-cluster-state storm-id callback)
              my-assignment (-> assignment :executor->node+port to-task->node+port (select-keys outbound-tasks))
              ;; we dont need a connection for the local tasks anymore
              needed-connections (->> my-assignment
                                      (filter-key (complement (-> worker :task-ids set)))

            ;; try using multipart messages ... first sort the tuples by the target node (without changing the local ordering)
            
            (fast-list-iter [[task ser-tuple] drainer]
              ;; TODO: consider write a batch of tuples here to every target worker  
              ;; group by node+port, do multipart send              
              (let [socket (get node+port->socket (get task->node+port task))]
                (msg/send socket task ser-tuple)
                ))))
        (.clear drainer)))))
