http://git-wip-us.apache.org/repos/asf/storm/blob/6b0da168/storm-core/test/clj/backtype/storm/nimbus_test.clj ---------------------------------------------------------------------- diff --cc storm-core/test/clj/backtype/storm/nimbus_test.clj index a3d1088,efdad25..151f3c7 --- a/storm-core/test/clj/backtype/storm/nimbus_test.clj +++ b/storm-core/test/clj/backtype/storm/nimbus_test.clj @@@ -15,13 -15,14 +15,16 @@@ ;; limitations under the License. (ns backtype.storm.nimbus-test (:use [clojure test]) + (:require [backtype.storm [util :as util]]) (:require [backtype.storm.daemon [nimbus :as nimbus]]) - (:import [backtype.storm.testing TestWordCounter TestWordSpout TestGlobalCount TestAggregatesCounter]) (:import [backtype.storm.scheduler INimbus]) + (:use [backtype.storm bootstrap testing zookeeper]) + (:import [backtype.storm.nimbus ILeaderElector NimbusInfo]) + (:use [backtype.storm.daemon common]) + (:import [backtype.storm.generated Credentials]) + (:use [backtype.storm bootstrap testing MockAutoCred]) - (:use [backtype.storm.daemon common]) + (:require [conjure.core]) (:use [conjure core]) ) @@@ -723,9 -762,74 +764,74 @@@ (bind executor-info (->> (storm-component->executor-info cluster "test") (map-val #(map executor-id->tasks %)))) (check-distribution (executor-info "1") [2 2 2 2 1 1 1 1]) - + ))) + + (defn check-for-collisions [state] + (log-message "Checking for collision") + (let [assignments (.assignments state nil)] + (log-message "Assignemts: " assignments) + (let [id->node->ports (into {} (for [id assignments + :let [executor->node+port (:executor->node+port (.assignment-info state id nil)) + node+ports (set (.values executor->node+port)) + node->ports (apply merge-with (fn [a b] (distinct (concat a b))) (for [[node port] node+ports] {node [port]}))]] + {id node->ports})) + _ (log-message "id->node->ports: " id->node->ports) + all-nodes (apply merge-with (fn [a b] + (let [ret (concat a b)] + (log-message "Can we combine " (pr-str a) " and " (pr-str b) " without collisions? " (apply distinct? ret) " => " (pr-str ret)) + (is (apply distinct? ret)) + (distinct ret))) + (.values id->node->ports))] + ))) + + (deftest test-rebalance-constrained-cluster + (with-simulated-time-local-cluster [cluster :supervisors 1 :ports-per-supervisor 4 + :daemon-conf {SUPERVISOR-ENABLE false + NIMBUS-MONITOR-FREQ-SECS 10 + TOPOLOGY-MESSAGE-TIMEOUT-SECS 30 + TOPOLOGY-ACKER-EXECUTORS 0}] + (letlocals + (bind topology (thrift/mk-topology + {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3)} + {})) + (bind topology2 (thrift/mk-topology + {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3)} + {})) + (bind topology3 (thrift/mk-topology + {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3)} + {})) + (bind state (:storm-cluster-state cluster)) + (submit-local-topology (:nimbus cluster) + "test" + {TOPOLOGY-WORKERS 3 + TOPOLOGY-MESSAGE-TIMEOUT-SECS 90} topology) + (submit-local-topology (:nimbus cluster) + "test2" + {TOPOLOGY-WORKERS 3 + TOPOLOGY-MESSAGE-TIMEOUT-SECS 90} topology2) + (submit-local-topology (:nimbus cluster) + "test3" + {TOPOLOGY-WORKERS 3 + TOPOLOGY-MESSAGE-TIMEOUT-SECS 90} topology3) + + (advance-cluster-time cluster 31) + + (check-for-collisions state) + (.rebalance (:nimbus cluster) "test" (doto (RebalanceOptions.) + (.set_num_workers 4) + (.set_wait_secs 0) + )) + + (advance-cluster-time cluster 11) + (check-for-collisions state) + + (advance-cluster-time cluster 30) + (check-for-collisions state) + ))) + + (deftest test-submit-invalid (with-simulated-time-local-cluster [cluster :daemon-conf {SUPERVISOR-ENABLE false @@@ -749,168 -855,314 +857,407 @@@ "test/aaa" {} topology))) - ))) + (bind topology (thrift/mk-topology + {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) + :parallelism-hint 16 + :conf {TOPOLOGY-TASKS 16})} + {})) + (bind state (:storm-cluster-state cluster)) + (is (thrown? InvalidTopologyException + (submit-local-topology (:nimbus cluster) + "test" + {TOPOLOGY-WORKERS 3} + topology))) + (bind topology (thrift/mk-topology + {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) + :parallelism-hint 5 + :conf {TOPOLOGY-TASKS 5})} + {})) + (is (thrown? InvalidTopologyException + (submit-local-topology (:nimbus cluster) + "test" + {TOPOLOGY-WORKERS 16} + topology))) + (is (nil? (submit-local-topology (:nimbus cluster) + "test" + {TOPOLOGY-WORKERS 8} + topology)))))) +(defnk mock-leader-elector [:is-leader true :leader-name "test-host" :leader-port 9999] + (let [leader-address (NimbusInfo. leader-name leader-port true)] + (reify ILeaderElector + (prepare [this conf] true) + (isLeader [this] is-leader) + (addToLeaderLockQueue [this] true) + (getLeader [this] leader-address) + (getAllNimbuses [this] `(leader-address)) + (close [this] true)))) + +(deftest test-cleans-corrupt + (with-inprocess-zookeeper zk-port + (with-local-tmp [nimbus-dir] + (stubbing [zk-leader-elector (mock-leader-elector)] + (letlocals + (bind conf (merge (read-storm-config) + {STORM-ZOOKEEPER-SERVERS ["localhost"] + STORM-CLUSTER-MODE "local" + STORM-ZOOKEEPER-PORT zk-port + STORM-LOCAL-DIR nimbus-dir})) + (bind cluster-state (cluster/mk-storm-cluster-state conf)) + (bind nimbus (nimbus/service-handler conf (nimbus/standalone-nimbus))) + (bind topology (thrift/mk-topology + {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3)} + {})) + (submit-local-topology nimbus "t1" {} topology) + (submit-local-topology nimbus "t2" {} topology) + (bind storm-id1 (get-storm-id cluster-state "t1")) + (bind storm-id2 (get-storm-id cluster-state "t2")) + (.shutdown nimbus) + (rmr (master-stormdist-root conf storm-id1)) + (bind nimbus (nimbus/service-handler conf (nimbus/standalone-nimbus))) + (is ( = #{storm-id2} (set (.active-storms cluster-state)))) + (.shutdown nimbus) + (.disconnect cluster-state) + ))))) + + (deftest test-cleans-corrupt (with-inprocess-zookeeper zk-port (with-local-tmp [nimbus-dir] - (letlocals - (bind conf (merge (read-storm-config) - {STORM-ZOOKEEPER-SERVERS ["localhost"] - STORM-CLUSTER-MODE "local" - STORM-ZOOKEEPER-PORT zk-port - STORM-LOCAL-DIR nimbus-dir})) - (bind cluster-state (cluster/mk-storm-cluster-state conf)) - (bind nimbus (nimbus/service-handler conf (nimbus/standalone-nimbus))) - (bind topology (thrift/mk-topology - {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3)} - {})) - (submit-local-topology nimbus "t1" {} topology) - (submit-local-topology nimbus "t2" {} topology) - (bind storm-id1 (get-storm-id cluster-state "t1")) - (bind storm-id2 (get-storm-id cluster-state "t2")) - (.shutdown nimbus) - (rmr (master-stormdist-root conf storm-id1)) - (bind nimbus (nimbus/service-handler conf (nimbus/standalone-nimbus))) - (is ( = #{storm-id2} (set (.active-storms cluster-state)))) - (.shutdown nimbus) - (.disconnect cluster-state) - )))) + (stubbing [zk-leader-elector (mock-leader-elector)] + (letlocals + (bind conf (merge (read-storm-config) + {STORM-ZOOKEEPER-SERVERS ["localhost"] + STORM-CLUSTER-MODE "local" + STORM-ZOOKEEPER-PORT zk-port + STORM-LOCAL-DIR nimbus-dir})) + (bind cluster-state (cluster/mk-storm-cluster-state conf)) + (bind nimbus (nimbus/service-handler conf (nimbus/standalone-nimbus))) + (bind topology (thrift/mk-topology + {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3)} + {})) + (submit-local-topology nimbus "t1" {} topology) + (submit-local-topology nimbus "t2" {} topology) + (bind storm-id1 (get-storm-id cluster-state "t1")) + (bind storm-id2 (get-storm-id cluster-state "t2")) + (.shutdown nimbus) + (rmr (master-stormdist-root conf storm-id1)) + (bind nimbus (nimbus/service-handler conf (nimbus/standalone-nimbus))) + (is ( = #{storm-id2} (set (.active-storms cluster-state)))) + (.shutdown nimbus) + (.disconnect cluster-state) + ))))) - (deftest test-no-overlapping-slots - ;; test that same node+port never appears across 2 assignments - ) + ;(deftest test-no-overlapping-slots + ; ;; test that same node+port never appears across 2 assignments + ; ) - (deftest test-stateless - ;; test that nimbus can die and restart without any problems - ) + ;(deftest test-stateless + ; ;; test that nimbus can die and restart without any problems + ; ) (deftest test-clean-inbox "Tests that the inbox correctly cleans jar files." (with-simulated-time - (with-local-tmp [dir-location] - (let [dir (File. dir-location) - mk-file (fn [name seconds-ago] - (let [f (File. (str dir-location "/" name)) - t (- (Time/currentTimeMillis) (* seconds-ago 1000))] - (FileUtils/touch f) - (.setLastModified f t))) - assert-files-in-dir (fn [compare-file-names] - (let [file-names (map #(.getName %) (file-seq dir))] - (is (= (sort compare-file-names) + (with-local-tmp [dir-location] + (let [dir (File. dir-location) + mk-file (fn [name seconds-ago] + (let [f (File. (str dir-location "/" name)) + t (- (Time/currentTimeMillis) (* seconds-ago 1000))] + (FileUtils/touch f) + (.setLastModified f t))) + assert-files-in-dir (fn [compare-file-names] + (let [file-names (map #(.getName %) (file-seq dir))] + (is (= (sort compare-file-names) (sort (filter #(.endsWith % ".jar") file-names)) ))))] - ;; Make three files a.jar, b.jar, c.jar. - ;; a and b are older than c and should be deleted first. - (advance-time-secs! 100) - (doseq [fs [["a.jar" 20] ["b.jar" 20] ["c.jar" 0]]] - (apply mk-file fs)) - (assert-files-in-dir ["a.jar" "b.jar" "c.jar"]) - (nimbus/clean-inbox dir-location 10) - (assert-files-in-dir ["c.jar"]) - ;; Cleanit again, c.jar should stay - (advance-time-secs! 5) - (nimbus/clean-inbox dir-location 10) - (assert-files-in-dir ["c.jar"]) - ;; Advance time, clean again, c.jar should be deleted. - (advance-time-secs! 5) - (nimbus/clean-inbox dir-location 10) - (assert-files-in-dir []) - )))) + ;; Make three files a.jar, b.jar, c.jar. + ;; a and b are older than c and should be deleted first. + (advance-time-secs! 100) + (doseq [fs [["a.jar" 20] ["b.jar" 20] ["c.jar" 0]]] + (apply mk-file fs)) + (assert-files-in-dir ["a.jar" "b.jar" "c.jar"]) + (nimbus/clean-inbox dir-location 10) + (assert-files-in-dir ["c.jar"]) + ;; Cleanit again, c.jar should stay + (advance-time-secs! 5) + (nimbus/clean-inbox dir-location 10) + (assert-files-in-dir ["c.jar"]) + ;; Advance time, clean again, c.jar should be deleted. + (advance-time-secs! 5) + (nimbus/clean-inbox dir-location 10) + (assert-files-in-dir []) + )))) + +(deftest test-leadership + "Tests that leader actions can only be performed by master and non leader fails to perform the same actions." + (with-inprocess-zookeeper zk-port + (with-local-tmp [nimbus-dir] + (stubbing [zk-leader-elector (mock-leader-elector)] + (letlocals + (bind conf (merge (read-storm-config) + {STORM-ZOOKEEPER-SERVERS ["localhost"] + STORM-CLUSTER-MODE "local" + STORM-ZOOKEEPER-PORT zk-port + STORM-LOCAL-DIR nimbus-dir})) + (bind cluster-state (cluster/mk-storm-cluster-state conf)) + (bind nimbus (nimbus/service-handler conf (nimbus/standalone-nimbus))) + (bind topology (thrift/mk-topology + {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3)} + {})) + + (stubbing [zk-leader-elector (mock-leader-elector :is-leader false)] + (letlocals + (bind non-leader-cluster-state (cluster/mk-storm-cluster-state conf)) + (bind non-leader-nimbus (nimbus/service-handler conf (nimbus/standalone-nimbus))) + + ;first we verify that the master nimbus can perform all actions, even with another nimbus present. + (submit-local-topology nimbus "t1" {} topology) + (.deactivate nimbus "t1") + (.activate nimbus "t1") + (.rebalance nimbus "t1" (RebalanceOptions.)) + (.killTopology nimbus "t1") + + ;now we verify that non master nimbus can not perform any of the actions. + (is (thrown? RuntimeException + (submit-local-topology non-leader-nimbus + "failing" + {} + topology))) + + (is (thrown? RuntimeException + (.killTopology non-leader-nimbus + "t1"))) + + (is (thrown? RuntimeException + (.activate non-leader-nimbus "t1"))) + + (is (thrown? RuntimeException + (.deactivate non-leader-nimbus "t1"))) + + (is (thrown? RuntimeException + (.rebalance non-leader-nimbus "t1" (RebalanceOptions.)))) + (.shutdown non-leader-nimbus) + (.disconnect non-leader-cluster-state) + )) + (.shutdown nimbus) + (.disconnect cluster-state)))))) + (deftest test-nimbus-iface-submitTopologyWithOpts-checks-authorization + (with-local-cluster [cluster + :daemon-conf {NIMBUS-AUTHORIZER + "backtype.storm.security.auth.authorizer.DenyAuthorizer"}] + (let [ + nimbus (:nimbus cluster) + topology (thrift/mk-topology {} {}) + ] + (is (thrown? AuthorizationException + (submit-local-topology-with-opts nimbus "mystorm" {} topology + (SubmitOptions. TopologyInitialStatus/INACTIVE)) + )) + ) + ) + ) + + (deftest test-nimbus-iface-methods-check-authorization + (with-local-cluster [cluster + :daemon-conf {NIMBUS-AUTHORIZER + "backtype.storm.security.auth.authorizer.DenyAuthorizer"}] + (let [ + nimbus (:nimbus cluster) + topology (thrift/mk-topology {} {}) + ] + ; Fake good authorization as part of setup. + (mocking [nimbus/check-authorization!] + (submit-local-topology-with-opts nimbus "test" {} topology + (SubmitOptions. TopologyInitialStatus/INACTIVE)) + ) + (stubbing [nimbus/storm-active? true] + (is (thrown? AuthorizationException + (.rebalance nimbus "test" (RebalanceOptions.)) + )) + ) + (is (thrown? AuthorizationException + (.activate nimbus "test") + )) + (is (thrown? AuthorizationException + (.deactivate nimbus "test") + )) + ) + ) + ) + + (deftest test-nimbus-check-authorization-params + (with-local-cluster [cluster + :daemon-conf {NIMBUS-AUTHORIZER "backtype.storm.security.auth.authorizer.NoopAuthorizer"}] + (let [nimbus (:nimbus cluster) + topology-name "test-nimbus-check-autho-params" + topology (thrift/mk-topology {} {})] + + (submit-local-topology-with-opts nimbus topology-name {} topology + (SubmitOptions. TopologyInitialStatus/INACTIVE)) + + (let [expected-name topology-name + expected-conf {TOPOLOGY-NAME expected-name + :foo :bar}] + + (testing "getTopologyConf calls check-authorization! with the correct parameters." + (let [expected-operation "getTopologyConf"] + (stubbing [nimbus/check-authorization! nil + nimbus/try-read-storm-conf expected-conf + util/to-json nil] + (try + (.getTopologyConf nimbus "fake-id") + (catch NotAliveException e) + (finally + (verify-first-call-args-for-indices + nimbus/check-authorization! + [1 2 3] expected-name expected-conf expected-operation) + (verify-first-call-args-for util/to-json expected-conf)))))) + + (testing "getTopology calls check-authorization! with the correct parameters." + (let [expected-operation "getTopology"] + (stubbing [nimbus/check-authorization! nil + nimbus/try-read-storm-conf expected-conf + nimbus/try-read-storm-topology nil + system-topology! nil] + (try + (.getTopology nimbus "fake-id") + (catch NotAliveException e) + (finally + (verify-first-call-args-for-indices + nimbus/check-authorization! + [1 2 3] expected-name expected-conf expected-operation) + (verify-first-call-args-for-indices + system-topology! [0] expected-conf)))))) + + (testing "getUserTopology calls check-authorization with the correct parameters." + (let [expected-operation "getUserTopology"] + (stubbing [nimbus/check-authorization! nil + nimbus/try-read-storm-conf expected-conf + nimbus/try-read-storm-topology nil] + (try + (.getUserTopology nimbus "fake-id") + (catch NotAliveException e) + (finally + (verify-first-call-args-for-indices + nimbus/check-authorization! + [1 2 3] expected-name expected-conf expected-operation) + (verify-first-call-args-for-indices + nimbus/try-read-storm-topology [0] expected-conf)))))))))) + + (deftest test-nimbus-iface-getTopology-methods-throw-correctly + (with-local-cluster [cluster] + (let [ + nimbus (:nimbus cluster) + id "bogus ID" + ] + (is (thrown? NotAliveException (.getTopology nimbus id))) + (try + (.getTopology nimbus id) + (catch NotAliveException e + (is (= id (.get_msg e))) + ) + ) + + (is (thrown? NotAliveException (.getTopologyConf nimbus id))) + (try (.getTopologyConf nimbus id) + (catch NotAliveException e + (is (= id (.get_msg e))) + ) + ) + + (is (thrown? NotAliveException (.getTopologyInfo nimbus id))) + (try (.getTopologyInfo nimbus id) + (catch NotAliveException e + (is (= id (.get_msg e))) + ) + ) + + (is (thrown? NotAliveException (.getUserTopology nimbus id))) + (try (.getUserTopology nimbus id) + (catch NotAliveException e + (is (= id (.get_msg e))) + ) + ) + ) + ) + ) + + (deftest test-nimbus-iface-getClusterInfo-filters-topos-without-bases + (with-local-cluster [cluster] + (let [ + nimbus (:nimbus cluster) + bogus-secs 42 + bogus-type "bogusType" + bogus-bases { + "1" nil + "2" {:launch-time-secs bogus-secs + :storm-name "id2-name" + :status {:type bogus-type}} + "3" nil + "4" {:launch-time-secs bogus-secs + :storm-name "id4-name" + :status {:type bogus-type}} + } + ] + (stubbing [topology-bases bogus-bases] + (let [topos (.get_topologies (.getClusterInfo nimbus))] + ; The number of topologies in the summary is correct. + (is (= (count + (filter (fn [b] (second b)) bogus-bases)) (count topos))) + ; Each topology present has a valid name. + (is (empty? + (filter (fn [t] (or (nil? t) (nil? (.get_name t)))) topos))) + ; The topologies are those with valid bases. + (is (empty? + (filter (fn [t] + (or + (nil? t) + (not (number? (read-string (.get_id t)))) + (odd? (read-string (.get_id t))) + )) topos))) + ) + ) + ) + ) + ) + + (deftest test-defserverfn-numbus-iface-instance + (test-nimbus-iface-submitTopologyWithOpts-checks-authorization) + (test-nimbus-iface-methods-check-authorization) + (test-nimbus-iface-getTopology-methods-throw-correctly) + (test-nimbus-iface-getClusterInfo-filters-topos-without-bases) + ) + + (deftest test-nimbus-data-acls + (testing "nimbus-data uses correct ACLs" + (let [scheme "digest" + digest "storm:thisisapoorpassword" + auth-conf {STORM-ZOOKEEPER-AUTH-SCHEME scheme + STORM-ZOOKEEPER-AUTH-PAYLOAD digest} + expected-acls nimbus/NIMBUS-ZK-ACLS + fake-inimbus (reify INimbus (getForcedScheduler [this] nil))] + (stubbing [mk-authorization-handler nil + cluster/mk-storm-cluster-state nil + nimbus/file-cache-map nil + uptime-computer nil + new-instance nil + mk-timer nil + nimbus/mk-scheduler nil] + (nimbus/nimbus-data auth-conf fake-inimbus) + (verify-call-times-for cluster/mk-storm-cluster-state 1) + (verify-first-call-args-for-indices cluster/mk-storm-cluster-state [2] + expected-acls))))) + + (deftest test-file-bogus-download + (with-local-cluster [cluster :daemon-conf {SUPERVISOR-ENABLE false TOPOLOGY-ACKER-EXECUTORS 0}] + (let [nimbus (:nimbus cluster)] + (is (thrown-cause? AuthorizationException (.beginFileDownload nimbus nil))) + (is (thrown-cause? AuthorizationException (.beginFileDownload nimbus ""))) + (is (thrown-cause? AuthorizationException (.beginFileDownload nimbus "/bogus-path/foo"))) + ))) + (deftest test-validate-topo-config-on-submit (with-local-cluster [cluster] (let [nimbus (:nimbus cluster)
http://git-wip-us.apache.org/repos/asf/storm/blob/6b0da168/storm-core/test/clj/backtype/storm/security/auth/auth_test.clj ---------------------------------------------------------------------- diff --cc storm-core/test/clj/backtype/storm/security/auth/auth_test.clj index ae79107,a01cac3..0e697d9 --- a/storm-core/test/clj/backtype/storm/security/auth/auth_test.clj +++ b/storm-core/test/clj/backtype/storm/security/auth/auth_test.clj @@@ -170,16 -328,17 +328,16 @@@ (let [a-port (available-port)] (with-server [a-port nil "backtype.storm.security.auth.authorizer.DenyAuthorizer" - "backtype.storm.security.auth.SimpleTransportPlugin"] + "backtype.storm.security.auth.SimpleTransportPlugin" nil] (let [storm-conf (merge (read-storm-config) {STORM-THRIFT-TRANSPORT-PLUGIN "backtype.storm.security.auth.SimpleTransportPlugin" - Config/NIMBUS_HOST "localhost" Config/NIMBUS_THRIFT_PORT a-port Config/NIMBUS_TASK_TIMEOUT_SECS nimbus-timeout}) - client (NimbusClient/getConfiguredClient storm-conf) + client (NimbusClient. storm-conf "localhost" a-port nimbus-timeout) nimbus_client (.getClient client)] (testing "(Negative authorization) Authorization plugin should reject client request" - (is (thrown? TTransportException - (.activate nimbus_client "security_auth_test_topology")))) + (is (thrown-cause? AuthorizationException + (.activate nimbus_client "security_auth_test_topology")))) (.close client))))) (deftest digest-authentication-test http://git-wip-us.apache.org/repos/asf/storm/blob/6b0da168/storm-core/test/clj/backtype/storm/utils_test.clj ---------------------------------------------------------------------- diff --cc storm-core/test/clj/backtype/storm/utils_test.clj index 86d971b,3b86540..2b0bbd5 --- a/storm-core/test/clj/backtype/storm/utils_test.clj +++ b/storm-core/test/clj/backtype/storm/utils_test.clj @@@ -46,9 -46,10 +46,9 @@@ (let [storm-conf (merge (read-storm-config) {STORM-THRIFT-TRANSPORT-PLUGIN "backtype.storm.security.auth.SimpleTransportPlugin" - Config/NIMBUS_HOST "" - Config/NIMBUS_THRIFT_PORT 65535 - STORM-NIMBUS-RETRY-TIMES 0})] + Config/NIMBUS_THRIFT_PORT "" + })] - (is (thrown? RuntimeException + (is (thrown-cause? RuntimeException (NimbusClient/getConfiguredClient storm-conf))) ) )
