cleanup curator integration and don't close multiple times in tests

Nathan Marz · Nathan Marz · commit 0eb696b1701a · 2012-01-04T18:31:15.000-08:00
diff --git a/src/clj/backtype/storm/cluster.clj b/src/clj/backtype/storm/cluster.clj
@@ -23,26 +23,20 @@
 (defn mk-distributed-cluster-state [conf]
   (let [zk (zk/mk-client (mk-zk-connect-string (assoc conf STORM-ZOOKEEPER-ROOT "/")))]
     (zk/mkdirs zk (conf STORM-ZOOKEEPER-ROOT))
-    (if (.isStarted zk) 
-      (.close zk))
-    )
+    (.close zk))
   (let [callbacks (atom {})
         active (atom true)
         mk-zk #(zk/mk-client (mk-zk-connect-string conf)
                              (conf STORM-ZOOKEEPER-SESSION-TIMEOUT)
                              %)
-        zk (atom nil)
-        watcher (fn this [state type path]
+        zk (mk-zk (fn [state type path]
                     (when @active
                       (when-not (= :connected state)
-                        (log-message "Zookeeper disconnected. Attempting to reconnect")
-                       ; (reset! zk (mk-zk this)) ;we don't need to reset it anymore,curator will take care of it.
-                        )
+                        (log-warn "Received event " state ":" type ":" path " with disconnected Zookeeper."))
                       (when-not (= :none type)
                         (doseq [callback (vals @callbacks)]                          
                           (callback type path))))
-                      )]
-    (reset! zk (mk-zk watcher))
+                      ))]
     (reify
      ClusterState
      (register [this callback]
@@ -53,39 +47,38 @@
      (unregister [this id]
                  (swap! callbacks dissoc id))
      (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
-                           (zk/create-node @zk path data :ephemeral)
+                         (zk/mkdirs zk (parent-path path))
+                         (if (zk/exists zk path false)
+                           (zk/set-data zk path data) ; should verify that it's ephemeral
+                           (zk/create-node zk path data :ephemeral)
                            ))
      
      (set-data [this path data]
                ;; note: this does not turn off any existing watches
-               (if (zk/exists @zk path false)
-                 (zk/set-data @zk path data)
+               (if (zk/exists zk path false)
+                 (zk/set-data zk path data)
                  (do
-                   (zk/mkdirs @zk (parent-path path))
-                   (zk/create-node @zk path data :persistent)
+                   (zk/mkdirs zk (parent-path path))
+                   (zk/create-node zk path data :persistent)
                    )))
      
      (delete-node [this path]
-                  (zk/delete-recursive @zk path)
+                  (zk/delete-recursive zk path)
                   )
      
      (get-data [this path watch?]
-               (zk/get-data @zk path watch?)
+               (zk/get-data zk path watch?)
                )
      
      (get-children [this path watch?]
-                   (zk/get-children @zk path watch?))
+                   (zk/get-children zk path watch?))
      
      (mkdirs [this path]
-             (zk/mkdirs @zk path))
+             (zk/mkdirs zk path))
      
      (close [this]
             (reset! active false)
-            (if (.isStarted @zk)
-              (.close @zk)))
+            (.close zk))
      )))
 
 (defprotocol StormClusterState
diff --git a/src/clj/backtype/storm/log.clj b/src/clj/backtype/storm/log.clj
@@ -12,3 +12,6 @@
 
 (defmacro log-warn-error [e & args]
   `(log/warn (str ~@args) ~e))
+
+(defmacro log-warn [& args]
+  `(log/warn (str ~@args)))
diff --git a/src/clj/backtype/storm/zookeeper.clj b/src/clj/backtype/storm/zookeeper.clj
@@ -35,7 +35,7 @@
                session-timeout
                15000
                ;;TODO: make retry times could be configured.
-               (RetryNTimes. 5 1000))]   
+               (RetryNTimes. 5 1000))]
       (.. fk (getCuratorListenable) (addListener (reify CuratorListener
                                                    (^void eventReceived [this ^CuratorFramework _fk ^CuratorEvent e]
                                                                         (let [^WatchedEvent event (.getWatchedEvent e)]
diff --git a/test/clj/backtype/storm/cluster_test.clj b/test/clj/backtype/storm/cluster_test.clj
@@ -68,9 +68,7 @@
       (is (Arrays/equals (barr 1) (.get-data state2 "/a" false)))
       (.close state1)
       (is (= nil (.get-data state2 "/a" false)))
-      (.close state1)
       (.close state2)
-      (.close state3)
       )))
 
 (defn mk-callback-tester []
@@ -224,7 +222,6 @@
       (.disconnect state2)
       (is (= #{"1"} (set (.supervisors state1 nil))))
       (.disconnect state1)
-      (.disconnect state2)
       )))
 
 (deftest test-storm-state-callbacks

Original file line number	Diff line number	Diff line change
`@@ -35,7 +35,7 @@`
`35`	`35`	`session-timeout`
`36`	`36`	`15000`
`37`	`37`	`;;TODO: make retry times could be configured.`
`38`		`- (RetryNTimes. 5 1000))]`
	`38`	`+ (RetryNTimes. 5 1000))]`
`39`	`39`	`(.. fk (getCuratorListenable) (addListener (reify CuratorListener`
`40`	`40`	`(^void eventReceived [this ^CuratorFramework _fk ^CuratorEvent e]`
`41`	`41`	`(let [^WatchedEvent event (.getWatchedEvent e)]`