             )))
 
       (^String getNimbusConf [this]
+        (ensure-leadership leader-elections)
         (to-json (:conf nimbus)))
 
       (^String getTopologyConf [this ^String id]
+        (ensure-leadership leader-elections)
         (to-json (read-storm-conf storage conf id)))
 
       (^StormTopology getTopology [this ^String id]
+        (ensure-leadership leader-elections)
         (system-topology! (read-storm-conf storage conf id) (read-storm-topology storage id)))
 
       (^StormTopology getUserTopology [this ^String id]
+        (ensure-leadership leader-elections)
         (read-storm-topology storage id))
 
       (^ClusterSummary getClusterInfo [this]
+        (ensure-leadership leader-elections)
         (let [storm-cluster-state (:storm-cluster-state nimbus)
               supervisor-infos (all-supervisor-info storm-cluster-state)
               ;; TODO: need to get the port info about supervisors...

          ))

      (^void uploadChunk [this ^String location ^ByteBuffer chunk]
        (let [uploaders (:uploaders nimbus)
              ^WritableByteChannel channel (.get uploaders location)]
          (when-not channel

      
      (^void submitTopology
        [this ^String storm-name ^String uploadedJarLocation ^String serializedConf ^StormTopology topology]
        (ensure-leadership leader-elections)
        (.submitTopologyWithOpts this storm-name uploadedJarLocation serializedConf topology
                                 (SubmitOptions. TopologyInitialStatus/ACTIVE)))
      
      (^void killTopology [this ^String name]
        (ensure-leadership leader-elections)
        (.killTopologyWithOpts this name (KillOptions.)))

      (^void killTopologyWithOpts [this ^String storm-name ^KillOptions options]
        (ensure-leadership leader-elections)
        (check-storm-active! nimbus storm-name true)
        (let [wait-amt (if (.is_set_wait_secs options)
                         (.get_wait_secs options)                         
