      (set-ephemeral-node [this path data]
                          (zk/mkdirs zk (parent-path path))
                          (if (zk/exists zk path false)
-                           (zk/set-data zk path data) ; should verify that it's ephemeral
+                           (try-cause
+                             (zk/set-data zk path data) ; should verify that it's ephemeral
+                             (catch KeeperException$NoNodeException e
+                               (log-warn-error e "Ephemeral node disappeared between checking for existing and setting data")
+                               (zk/create-node zk path data :ephemeral)
+                               ))
                            (zk/create-node zk path data :ephemeral)
                            ))
      

(ns backtype.storm.cluster
  (:import [org.apache.zookeeper.data Stat])
  (:import [org.apache.zookeeper KeeperException])
  (:import [backtype.storm.utils Utils])
  (:use [backtype.storm util log config])
  (:use [clojure.contrib.core :only [dissoc-in]])

        )

      (task-heartbeat! [this storm-id task-id info]
        (set-data cluster-state (taskbeat-path storm-id task-id) (Utils/serialize info))
        )

      (remove-task-heartbeat! [this storm-id task-id]
