       (^void submitTopologyWithOpts
         [this ^String storm-name ^String uploadedJarLocation ^String serializedConf ^StormTopology topology
          ^SubmitOptions submitOptions]
+        (ensure-leadership leader-elections)
         (try
           (assert (not-nil? submitOptions))
           (validate-topology-name! storm-name)

          ))
      
      (^TopologyInfo getTopologyInfo [this ^String storm-id]
        (let [storm-cluster-state (:storm-cluster-state nimbus)
              storm-topology (read-storm-topology storage storm-id)
              storm-conf (read-storm-conf storage conf storm-id)

          ))

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