Repository: storm Updated Branches: refs/heads/master 795edec74 -> 7f14dfd88
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/test/clj/backtype/storm/messaging/netty_unit_test.clj ---------------------------------------------------------------------- diff --git a/storm-core/test/clj/backtype/storm/messaging/netty_unit_test.clj b/storm-core/test/clj/backtype/storm/messaging/netty_unit_test.clj index f7d3802..aa806a7 100644 --- a/storm-core/test/clj/backtype/storm/messaging/netty_unit_test.clj +++ b/storm-core/test/clj/backtype/storm/messaging/netty_unit_test.clj @@ -16,6 +16,7 @@ (ns backtype.storm.messaging.netty-unit-test (:use [clojure test]) (:import [backtype.storm.messaging TransportFactory]) + (:import [backtype.storm.utils Utils]) (:use [backtype.storm testing util config log]) (:use [backtype.storm.daemon.worker :only [is-connection-ready]]) (:import [java.util ArrayList])) @@ -33,20 +34,37 @@ (do (log-message "Waiting until all Netty connections are ready...") (wait-until-ready connections 0))) ([connections waited-ms] - (let [interval-ms 10 - max-wait-ms 5000] - (if-not (every? is-connection-ready connections) - (if (<= waited-ms max-wait-ms) - (do - (Thread/sleep interval-ms) - (wait-until-ready connections (+ waited-ms interval-ms))) - (throw (RuntimeException. (str "Netty connections were not ready within " max-wait-ms " ms")))) - (log-message "All Netty connections are ready"))))) + (loop [connections connections waited-ms waited-ms] + (let [interval-ms 10 + max-wait-ms 5000] + (if-not (every? is-connection-ready connections) + (if (<= waited-ms max-wait-ms) + (do + (Thread/sleep interval-ms) + (recur connections (+ waited-ms interval-ms))) + (throw (RuntimeException. (str "Netty connections were not ready within " max-wait-ms " ms")))) + (log-message "All Netty connections are ready")))))) -(deftest test-basic - (log-message "Should send and receive a basic message") + +(defn- test-basic-fn [storm-conf] + (log-message "1. Should send and receive a basic message") (let [req_msg (String. "0123456789abcdefghijklmnopqrstuvwxyz") - storm-conf {STORM-MESSAGING-TRANSPORT "backtype.storm.messaging.netty.Context" + context (TransportFactory/makeContext storm-conf) + port (available-port 6700) + server (.bind context nil port) + client (.connect context nil "localhost" port) + _ (wait-until-ready [server client]) + _ (.send client task (.getBytes req_msg)) + iter (.recv server 0 0) + resp (.next iter)] + (is (= task (.task resp))) + (is (= req_msg (String. (.message resp)))) + (.close client) + (.close server) + (.term context))) + +(deftest test-basic + (let [storm-conf {STORM-MESSAGING-TRANSPORT "backtype.storm.messaging.netty.Context" STORM-MESSAGING-NETTY-AUTHENTICATION false STORM-MESSAGING-NETTY-BUFFER-SIZE 1024 STORM-MESSAGING-NETTY-MAX-RETRIES 10 @@ -54,33 +72,64 @@ STORM-MESSAGING-NETTY-MAX-SLEEP-MS 5000 STORM-MESSAGING-NETTY-SERVER-WORKER-THREADS 1 STORM-MESSAGING-NETTY-CLIENT-WORKER-THREADS 1 - } + TOPOLOGY-KRYO-FACTORY "backtype.storm.serialization.DefaultKryoFactory" + TOPOLOGY-TUPLE-SERIALIZER "backtype.storm.serialization.types.ListDelegateSerializer" + TOPOLOGY-FALL-BACK-ON-JAVA-SERIALIZATION false + TOPOLOGY-SKIP-MISSING-KRYO-REGISTRATIONS false} + storm-conf-sasl (assoc storm-conf + STORM-MESSAGING-NETTY-AUTHENTICATION true + TOPOLOGY-NAME "topo1-netty-sasl" + STORM-ZOOKEEPER-TOPOLOGY-AUTH-PAYLOAD (str (Utils/secureRandomLong) ":" (Utils/secureRandomLong)))] + (test-basic-fn storm-conf) ;; test with sasl authentication disabled + (test-basic-fn storm-conf-sasl))) ;; test with sasl authentication enabled + +(defn- test-load-fn [storm-conf] + (log-message "2 test load") + (let [req_msg (String. "0123456789abcdefghijklmnopqrstuvwxyz") context (TransportFactory/makeContext storm-conf) + port (available-port 6700) server (.bind context nil port) client (.connect context nil "localhost" port) _ (wait-until-ready [server client]) _ (.send client task (.getBytes req_msg)) iter (.recv server 0 0) - resp (.next iter)] + resp (.next iter) + _ (.sendLoadMetrics server {(int 1) 0.0 (int 2) 1.0}) + _ (while-timeout 5000 (empty? (.getLoad client [(int 1) (int 2)])) (Thread/sleep 10)) + load (.getLoad client [(int 1) (int 2)])] + (is (= 0.0 (.getBoltLoad (.get load (int 1))))) + (is (= 1.0 (.getBoltLoad (.get load (int 2))))) (is (= task (.task resp))) (is (= req_msg (String. (.message resp)))) (.close client) (.close server) (.term context))) -(deftest test-large-msg - (log-message "Should send and receive a large message") - (let [req_msg (apply str (repeat 2048000 'c')) - storm-conf {STORM-MESSAGING-TRANSPORT "backtype.storm.messaging.netty.Context" +(deftest test-load + (let [storm-conf {STORM-MESSAGING-TRANSPORT "backtype.storm.messaging.netty.Context" STORM-MESSAGING-NETTY-AUTHENTICATION false - STORM-MESSAGING-NETTY-BUFFER-SIZE 102400 + STORM-MESSAGING-NETTY-BUFFER-SIZE 1024 STORM-MESSAGING-NETTY-MAX-RETRIES 10 STORM-MESSAGING-NETTY-MIN-SLEEP-MS 1000 STORM-MESSAGING-NETTY-MAX-SLEEP-MS 5000 STORM-MESSAGING-NETTY-SERVER-WORKER-THREADS 1 STORM-MESSAGING-NETTY-CLIENT-WORKER-THREADS 1 - } + TOPOLOGY-KRYO-FACTORY "backtype.storm.serialization.DefaultKryoFactory" + TOPOLOGY-TUPLE-SERIALIZER "backtype.storm.serialization.types.ListDelegateSerializer" + TOPOLOGY-FALL-BACK-ON-JAVA-SERIALIZATION false + TOPOLOGY-SKIP-MISSING-KRYO-REGISTRATIONS false} + storm-conf-sasl (assoc storm-conf + STORM-MESSAGING-NETTY-AUTHENTICATION true + TOPOLOGY-NAME "topo1-netty-sasl" + STORM-ZOOKEEPER-TOPOLOGY-AUTH-PAYLOAD (str (Utils/secureRandomLong) ":" (Utils/secureRandomLong)))] + (test-load-fn storm-conf) ;; test with sasl authentication disabled + (test-load-fn storm-conf-sasl))) ;; test with sasl authentication enabled + +(defn test-large-msg-fn [storm-conf] + (log-message "3 Should send and receive a large message") + (let [req_msg (apply str (repeat 2048000 'c')) context (TransportFactory/makeContext storm-conf) + port (available-port 6700) server (.bind context nil port) client (.connect context nil "localhost" port) _ (wait-until-ready [server client]) @@ -93,20 +142,77 @@ (.close server) (.term context))) +(deftest test-large-msg + (let [storm-conf {STORM-MESSAGING-TRANSPORT "backtype.storm.messaging.netty.Context" + STORM-MESSAGING-NETTY-AUTHENTICATION false + STORM-MESSAGING-NETTY-BUFFER-SIZE 102400 + STORM-MESSAGING-NETTY-MAX-RETRIES 10 + STORM-MESSAGING-NETTY-MIN-SLEEP-MS 1000 + STORM-MESSAGING-NETTY-MAX-SLEEP-MS 5000 + STORM-MESSAGING-NETTY-SERVER-WORKER-THREADS 1 + STORM-MESSAGING-NETTY-CLIENT-WORKER-THREADS 1 + TOPOLOGY-KRYO-FACTORY "backtype.storm.serialization.DefaultKryoFactory" + TOPOLOGY-TUPLE-SERIALIZER "backtype.storm.serialization.types.ListDelegateSerializer" + TOPOLOGY-FALL-BACK-ON-JAVA-SERIALIZATION false + TOPOLOGY-SKIP-MISSING-KRYO-REGISTRATIONS false} + storm-conf-sasl (assoc storm-conf + STORM-MESSAGING-NETTY-AUTHENTICATION true + TOPOLOGY-NAME "topo1-netty-sasl" + STORM-ZOOKEEPER-TOPOLOGY-AUTH-PAYLOAD (str (Utils/secureRandomLong) ":" (Utils/secureRandomLong)))] + (test-large-msg-fn storm-conf) ;; test with sasl authentication disabled + (test-large-msg-fn storm-conf-sasl))) ;; test with sasl authentication enabled -(deftest test-batch - (let [num-messages 100000 - storm-conf {STORM-MESSAGING-TRANSPORT "backtype.storm.messaging.netty.Context" +(defn- test-server-delayed-fn [storm-conf] + (log-message "4. test server delayed") + (let [req_msg (String. "0123456789abcdefghijklmnopqrstuvwxyz") + context (TransportFactory/makeContext storm-conf) + port (available-port 6700) + client (.connect context nil "localhost" port) + + server (Thread. + (fn [] + (Thread/sleep 1000) + (let [server (.bind context nil port) + iter (.recv server 0 0) + resp (.next iter)] + (is (= task (.task resp))) + (is (= req_msg (String. (.message resp)))) + (.close server))))] + (.start server) + (wait-until-ready [server client]) + (.send client task (.getBytes req_msg)) + + (.join server) + (.close client) + (.term context))) + +(deftest test-server-delayed + (let [storm-conf {STORM-MESSAGING-TRANSPORT "backtype.storm.messaging.netty.Context" STORM-MESSAGING-NETTY-AUTHENTICATION false - STORM-MESSAGING-NETTY-BUFFER-SIZE 1024000 + STORM-MESSAGING-NETTY-BUFFER-SIZE 1024 STORM-MESSAGING-NETTY-MAX-RETRIES 10 STORM-MESSAGING-NETTY-MIN-SLEEP-MS 1000 STORM-MESSAGING-NETTY-MAX-SLEEP-MS 5000 STORM-MESSAGING-NETTY-SERVER-WORKER-THREADS 1 STORM-MESSAGING-NETTY-CLIENT-WORKER-THREADS 1 - } + TOPOLOGY-KRYO-FACTORY "backtype.storm.serialization.DefaultKryoFactory" + TOPOLOGY-TUPLE-SERIALIZER "backtype.storm.serialization.types.ListDelegateSerializer" + TOPOLOGY-FALL-BACK-ON-JAVA-SERIALIZATION false + TOPOLOGY-SKIP-MISSING-KRYO-REGISTRATIONS false} + storm-conf-sasl (assoc storm-conf + STORM-MESSAGING-NETTY-AUTHENTICATION true + TOPOLOGY-NAME "topo1-netty-sasl" + STORM-ZOOKEEPER-TOPOLOGY-AUTH-PAYLOAD (str (Utils/secureRandomLong) ":" (Utils/secureRandomLong)))] + (test-server-delayed-fn storm-conf) ;; test with sasl authentication disabled + (test-server-delayed-fn storm-conf-sasl))) ;; test with sasl authentication enabled + + +(defn- test-batch-fn [storm-conf] + (log-message "5. test batch") + (let [num-messages 100000 _ (log-message "Should send and receive many messages (testing with " num-messages " messages)") context (TransportFactory/makeContext storm-conf) + port (available-port 6700) server (.bind context nil port) client (.connect context nil "localhost" port) _ (wait-until-ready [server client])] @@ -130,4 +236,78 @@ (.close client) (.close server) - (.term context))) + (.term context)) + +(deftest test-batch + (let [storm-conf {STORM-MESSAGING-TRANSPORT "backtype.storm.messaging.netty.Context" + STORM-MESSAGING-NETTY-AUTHENTICATION false + STORM-MESSAGING-NETTY-BUFFER-SIZE 1024000 + STORM-MESSAGING-NETTY-MAX-RETRIES 10 + STORM-MESSAGING-NETTY-MIN-SLEEP-MS 1000 + STORM-MESSAGING-NETTY-MAX-SLEEP-MS 5000 + STORM-MESSAGING-NETTY-SERVER-WORKER-THREADS 1 + STORM-MESSAGING-NETTY-CLIENT-WORKER-THREADS 1 + TOPOLOGY-KRYO-FACTORY "backtype.storm.serialization.DefaultKryoFactory" + TOPOLOGY-TUPLE-SERIALIZER "backtype.storm.serialization.types.ListDelegateSerializer" + TOPOLOGY-FALL-BACK-ON-JAVA-SERIALIZATION false + TOPOLOGY-SKIP-MISSING-KRYO-REGISTRATIONS false} + storm-conf-sasl (assoc storm-conf + STORM-MESSAGING-NETTY-AUTHENTICATION true + TOPOLOGY-NAME "topo1-netty-sasl" + STORM-ZOOKEEPER-TOPOLOGY-AUTH-PAYLOAD (str (Utils/secureRandomLong) ":" (Utils/secureRandomLong)))] + (test-batch-fn storm-conf) ;; test with sasl authentication disabled + (test-batch-fn storm-conf-sasl))) ;; test with sasl authentication enabled +) + +(defn- test-server-always-reconnects-fn [storm-conf] + (log-message "6. test server always reconnects") + (let [req_msg (String. "0123456789abcdefghijklmnopqrstuvwxyz") + storm-conf {STORM-MESSAGING-TRANSPORT "backtype.storm.messaging.netty.Context" + STORM-MESSAGING-NETTY-AUTHENTICATION false + STORM-MESSAGING-NETTY-BUFFER-SIZE 1024 + STORM-MESSAGING-NETTY-MAX-RETRIES 2 + STORM-MESSAGING-NETTY-MIN-SLEEP-MS 10 + STORM-MESSAGING-NETTY-MAX-SLEEP-MS 50 + STORM-MESSAGING-NETTY-SERVER-WORKER-THREADS 1 + STORM-MESSAGING-NETTY-CLIENT-WORKER-THREADS 1 + TOPOLOGY-KRYO-FACTORY "backtype.storm.serialization.DefaultKryoFactory" + TOPOLOGY-TUPLE-SERIALIZER "backtype.storm.serialization.types.ListDelegateSerializer" + TOPOLOGY-FALL-BACK-ON-JAVA-SERIALIZATION false + TOPOLOGY-SKIP-MISSING-KRYO-REGISTRATIONS false} + context (TransportFactory/makeContext storm-conf) + port (available-port 6700) + client (.connect context nil "localhost" port) + _ (.send client task (.getBytes req_msg)) + server (.bind context nil port) + _ (wait-until-ready [server client]) + _ (.send client task (.getBytes req_msg)) + iter (future (.recv server 0 0)) + resp (deref iter 5000 nil) + resp-val (if resp (.next resp) nil)] + (is resp-val) + (when resp-val + (is (= task (.task resp-val))) + (is (= req_msg (String. (.message resp-val))))) + (.close client) + (.close server) + (.term context))) + +(deftest test-server-always-reconnects + (let [storm-conf {STORM-MESSAGING-TRANSPORT "backtype.storm.messaging.netty.Context" + STORM-MESSAGING-NETTY-AUTHENTICATION false + STORM-MESSAGING-NETTY-BUFFER-SIZE 1024 + STORM-MESSAGING-NETTY-MAX-RETRIES 2 + STORM-MESSAGING-NETTY-MIN-SLEEP-MS 10 + STORM-MESSAGING-NETTY-MAX-SLEEP-MS 50 + STORM-MESSAGING-NETTY-SERVER-WORKER-THREADS 1 + STORM-MESSAGING-NETTY-CLIENT-WORKER-THREADS 1 + TOPOLOGY-KRYO-FACTORY "backtype.storm.serialization.DefaultKryoFactory" + TOPOLOGY-TUPLE-SERIALIZER "backtype.storm.serialization.types.ListDelegateSerializer" + TOPOLOGY-FALL-BACK-ON-JAVA-SERIALIZATION false + TOPOLOGY-SKIP-MISSING-KRYO-REGISTRATIONS false} + storm-conf-sasl (assoc storm-conf + STORM-MESSAGING-NETTY-AUTHENTICATION true + TOPOLOGY-NAME "topo1-netty-sasl" + STORM-ZOOKEEPER-TOPOLOGY-AUTH-PAYLOAD (str (Utils/secureRandomLong) ":" (Utils/secureRandomLong)))] + (test-server-always-reconnects-fn storm-conf) ;; test with sasl authentication disabled + (test-server-always-reconnects-fn storm-conf-sasl))) ;; test with sasl authentication enabled http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/test/clj/backtype/storm/messaging_test.clj ---------------------------------------------------------------------- diff --git a/storm-core/test/clj/backtype/storm/messaging_test.clj b/storm-core/test/clj/backtype/storm/messaging_test.clj index 53cd555..89d8586 100644 --- a/storm-core/test/clj/backtype/storm/messaging_test.clj +++ b/storm-core/test/clj/backtype/storm/messaging_test.clj @@ -51,8 +51,7 @@ ["a"] ["b"] ]} )] - (is (ms= (apply concat (repeat 6 [[1] [2] [3] [4]])) - (read-tuples results "2"))))))) + (is (= (* 6 4) (.size (read-tuples results "2")))))))) (extend-type TestEventLogSpout CompletableSpout
