You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2016/02/18 17:12:38 UTC
[1/5] storm git commit: backport thrift.clj to Thrift.java
Repository: storm
Updated Branches:
refs/heads/master bb60f11c0 -> d187a20e2
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/test/clj/org/apache/storm/nimbus_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/org/apache/storm/nimbus_test.clj b/storm-core/test/clj/org/apache/storm/nimbus_test.clj
index 70cb885..42a0374 100644
--- a/storm-core/test/clj/org/apache/storm/nimbus_test.clj
+++ b/storm-core/test/clj/org/apache/storm/nimbus_test.clj
@@ -20,7 +20,9 @@
(:require [org.apache.storm [converter :as converter]])
(:import [org.apache.storm.testing TestWordCounter TestWordSpout TestGlobalCount
TestAggregatesCounter TestPlannerSpout TestPlannerBolt]
- [org.apache.storm.nimbus InMemoryTopologyActionNotifier])
+ [org.apache.storm.nimbus InMemoryTopologyActionNotifier]
+ [org.apache.storm.generated GlobalStreamId]
+ [org.apache.storm Thrift])
(:import [org.apache.storm.testing.staticmocking MockedZookeeper])
(:import [org.apache.storm.scheduler INimbus])
(:import [org.apache.storm.nimbus ILeaderElector NimbusInfo])
@@ -38,9 +40,7 @@
(:use [org.apache.storm testing MockAutoCred util config log timer zookeeper])
(:use [org.apache.storm.daemon common])
(:require [conjure.core])
- (:require [org.apache.storm
- [thrift :as thrift]
- [cluster :as cluster]])
+ (:require [org.apache.storm [cluster :as cluster]])
(:use [conjure core]))
(defn- from-json
@@ -211,16 +211,34 @@
:daemon-conf {SUPERVISOR-ENABLE false TOPOLOGY-ACKER-EXECUTORS 0 TOPOLOGY-EVENTLOGGER-EXECUTORS 0}]
(let [state (:storm-cluster-state cluster)
nimbus (:nimbus cluster)
- topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. false) :parallelism-hint 3)}
- {"2" (thrift/mk-bolt-spec {"1" :none} (TestPlannerBolt.) :parallelism-hint 4)
- "3" (thrift/mk-bolt-spec {"2" :none} (TestPlannerBolt.))})
- topology2 (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 12)}
- {"2" (thrift/mk-bolt-spec {"1" :none} (TestPlannerBolt.) :parallelism-hint 6)
- "3" (thrift/mk-bolt-spec {"1" :global} (TestPlannerBolt.) :parallelism-hint 8)
- "4" (thrift/mk-bolt-spec {"1" :global "2" :none} (TestPlannerBolt.) :parallelism-hint 4)}
- )
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. false) (Integer. 3))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareNoneGrouping)}
+ (TestPlannerBolt.) (Integer. 4))
+ "3" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "2" nil)
+ (Thrift/prepareNoneGrouping)}
+ (TestPlannerBolt.))})
+ topology2 (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 12))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareNoneGrouping)}
+ (TestPlannerBolt.) (Integer. 6))
+ "3" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareGlobalGrouping)}
+ (TestPlannerBolt.) (Integer. 8))
+ "4" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareGlobalGrouping)
+ (Utils/getGlobalStreamId "2" nil)
+ (Thrift/prepareNoneGrouping)}
+ (TestPlannerBolt.) (Integer. 4))})
_ (submit-local-topology nimbus "mystorm" {TOPOLOGY-WORKERS 4} topology)
_ (advance-cluster-time cluster 11)
task-info (storm-component->task-info cluster "mystorm")]
@@ -278,10 +296,17 @@
topology-name "test-auto-cred-storm"
submitOptions (SubmitOptions. TopologyInitialStatus/INACTIVE)
- (.set_creds submitOptions (Credentials. (HashMap.)))
- topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. false) :parallelism-hint 3)}
- {"2" (thrift/mk-bolt-spec {"1" :none} (TestPlannerBolt.) :parallelism-hint 4)
- "3" (thrift/mk-bolt-spec {"2" :none} (TestPlannerBolt.))})
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. false) (Integer. 3))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareNoneGrouping)}
+ (TestPlannerBolt.) (Integer. 4))
+ "3" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "2" nil)
+ (Thrift/prepareNoneGrouping)}
+ (TestPlannerBolt.))})
_ (submit-local-topology-with-opts nimbus topology-name {TOPOLOGY-WORKERS 4
TOPOLOGY-AUTO-CREDENTIALS (list "org.apache.storm.MockAutoCred")
} topology submitOptions)
@@ -320,10 +345,17 @@
(letlocals
(bind state (:storm-cluster-state cluster))
(bind nimbus (:nimbus cluster))
- (bind topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. false) :parallelism-hint 3)}
- {"2" (thrift/mk-bolt-spec {"1" :none} (TestPlannerBolt.) :parallelism-hint 5)
- "3" (thrift/mk-bolt-spec {"2" :none} (TestPlannerBolt.))}))
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. false) (Integer. 3))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareNoneGrouping)}
+ (TestPlannerBolt.) (Integer. 5))
+ "3" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "2" nil)
+ (Thrift/prepareNoneGrouping)}
+ (TestPlannerBolt.))}))
(submit-local-topology nimbus "noniso" {TOPOLOGY-WORKERS 4} topology)
(advance-cluster-time cluster 11)
@@ -365,10 +397,20 @@
(with-simulated-time-local-cluster [cluster :daemon-conf {SUPERVISOR-ENABLE false TOPOLOGY-ACKER-EXECUTORS 0 TOPOLOGY-EVENTLOGGER-EXECUTORS 0}]
(let [state (:storm-cluster-state cluster)
nimbus (:nimbus cluster)
- topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. false) :parallelism-hint 3 :conf {TOPOLOGY-TASKS 0})}
- {"2" (thrift/mk-bolt-spec {"1" :none} (TestPlannerBolt.) :parallelism-hint 1 :conf {TOPOLOGY-TASKS 2})
- "3" (thrift/mk-bolt-spec {"2" :none} (TestPlannerBolt.) :conf {TOPOLOGY-TASKS 5})})
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. false) (Integer. 3)
+ {TOPOLOGY-TASKS 0})}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareNoneGrouping)}
+ (TestPlannerBolt.) (Integer. 1)
+ {TOPOLOGY-TASKS 2})
+ "3" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "2" nil)
+ (Thrift/prepareNoneGrouping)}
+ (TestPlannerBolt.) nil
+ {TOPOLOGY-TASKS 5})})
_ (submit-local-topology nimbus "mystorm" {TOPOLOGY-WORKERS 4} topology)
_ (advance-cluster-time cluster 11)
task-info (storm-component->task-info cluster "mystorm")]
@@ -383,10 +425,19 @@
(deftest test-executor-assignments
(with-simulated-time-local-cluster[cluster :daemon-conf {SUPERVISOR-ENABLE false TOPOLOGY-ACKER-EXECUTORS 0 TOPOLOGY-EVENTLOGGER-EXECUTORS 0}]
(let [nimbus (:nimbus cluster)
- topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3 :conf {TOPOLOGY-TASKS 5})}
- {"2" (thrift/mk-bolt-spec {"1" :none} (TestPlannerBolt.) :parallelism-hint 8 :conf {TOPOLOGY-TASKS 2})
- "3" (thrift/mk-bolt-spec {"2" :none} (TestPlannerBolt.) :parallelism-hint 3)})
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 3)
+ {TOPOLOGY-TASKS 5})}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareNoneGrouping)}
+ (TestPlannerBolt.) (Integer. 8)
+ {TOPOLOGY-TASKS 2})
+ "3" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "2" nil)
+ (Thrift/prepareNoneGrouping)}
+ (TestPlannerBolt.) (Integer. 3))})
_ (submit-local-topology nimbus "mystorm" {TOPOLOGY-WORKERS 4} topology)
_ (advance-cluster-time cluster 11)
task-info (storm-component->task-info cluster "mystorm")
@@ -408,12 +459,21 @@
:daemon-conf {SUPERVISOR-ENABLE false TOPOLOGY-ACKER-EXECUTORS 0 TOPOLOGY-EVENTLOGGER-EXECUTORS 0}]
(let [state (:storm-cluster-state cluster)
nimbus (:nimbus cluster)
- topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 21)}
- {"2" (thrift/mk-bolt-spec {"1" :none} (TestPlannerBolt.) :parallelism-hint 9)
- "3" (thrift/mk-bolt-spec {"1" :none} (TestPlannerBolt.) :parallelism-hint 2)
- "4" (thrift/mk-bolt-spec {"1" :none} (TestPlannerBolt.) :parallelism-hint 10)}
- )
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 21))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareNoneGrouping)}
+ (TestPlannerBolt.) (Integer. 9))
+ "3" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareNoneGrouping)}
+ (TestPlannerBolt.) (Integer. 2))
+ "4" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareNoneGrouping)}
+ (TestPlannerBolt.) (Integer. 10))})
_ (submit-local-topology nimbus "test" {TOPOLOGY-WORKERS 7} topology)
_ (advance-cluster-time cluster 11)
task-info (storm-component->task-info cluster "test")]
@@ -436,10 +496,10 @@
(stubbing [nimbus/user-groups ["alice-group"]]
(letlocals
(bind conf (:daemon-conf cluster))
- (bind topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 4)}
- {}
- ))
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 4))}
+ {}))
(bind state (:storm-cluster-state cluster))
(submit-local-topology (:nimbus cluster) "test" {TOPOLOGY-MESSAGE-TIMEOUT-SECS 20, LOGS-USERS ["alice", (System/getProperty "user.name")]} topology)
(bind storm-id (get-storm-id state "test"))
@@ -532,10 +592,10 @@
TOPOLOGY-EVENTLOGGER-EXECUTORS 0}]
(letlocals
(bind conf (:daemon-conf cluster))
- (bind topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 14)}
- {}
- ))
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 14))}
+ {}))
(bind state (:storm-cluster-state cluster))
(submit-local-topology (:nimbus cluster) "test" {TOPOLOGY-MESSAGE-TIMEOUT-SECS 20} topology)
(bind storm-id (get-storm-id state "test"))
@@ -626,10 +686,10 @@
TOPOLOGY-EVENTLOGGER-EXECUTORS 0}]
(letlocals
(bind conf (:daemon-conf cluster))
- (bind topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 2)}
- {}
- ))
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 2))}
+ {}))
(bind state (:storm-cluster-state cluster))
(submit-local-topology (:nimbus cluster) "test" {TOPOLOGY-WORKERS 2} topology)
(advance-cluster-time cluster 11)
@@ -747,10 +807,10 @@
(add-supervisor cluster :ports 1 :id "a")
(add-supervisor cluster :ports 1 :id "b")
(bind conf (:daemon-conf cluster))
- (bind topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 2)}
- {}
- ))
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 2))}
+ {}))
(bind state (:storm-cluster-state cluster))
(submit-local-topology (:nimbus cluster) "test" {TOPOLOGY-WORKERS 2} topology)
(advance-cluster-time cluster 11)
@@ -803,8 +863,9 @@
TOPOLOGY-ACKER-EXECUTORS 0
TOPOLOGY-EVENTLOGGER-EXECUTORS 0}]
(letlocals
- (bind topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 9)}
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 9))}
{}))
(bind state (:storm-cluster-state cluster))
(submit-local-topology (:nimbus cluster) "test" {TOPOLOGY-WORKERS 4} topology) ; distribution should be 2, 2, 2, 3 ideally
@@ -852,8 +913,9 @@
TOPOLOGY-ACKER-EXECUTORS 0
TOPOLOGY-EVENTLOGGER-EXECUTORS 0}]
(letlocals
- (bind topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3)}
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 3))}
{}))
(bind state (:storm-cluster-state cluster))
(submit-local-topology (:nimbus cluster)
@@ -898,10 +960,10 @@
TOPOLOGY-ACKER-EXECUTORS 0
TOPOLOGY-EVENTLOGGER-EXECUTORS 0}]
(letlocals
- (bind topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true)
- :parallelism-hint 6
- :conf {TOPOLOGY-TASKS 12})}
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 6)
+ {TOPOLOGY-TASKS 12})}
{}))
(bind state (:storm-cluster-state cluster))
(submit-local-topology (:nimbus cluster)
@@ -976,14 +1038,17 @@
TOPOLOGY-ACKER-EXECUTORS 0
TOPOLOGY-EVENTLOGGER-EXECUTORS 0}]
(letlocals
- (bind topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3)}
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 3))}
{}))
- (bind topology2 (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3)}
+ (bind topology2 (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 3))}
{}))
- (bind topology3 (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3)}
+ (bind topology3 (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 3))}
{}))
(bind state (:storm-cluster-state cluster))
(submit-local-topology (:nimbus cluster)
@@ -1023,18 +1088,20 @@
NIMBUS-EXECUTORS-PER-TOPOLOGY 8
NIMBUS-SLOTS-PER-TOPOLOGY 8}]
(letlocals
- (bind topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 1 :conf {TOPOLOGY-TASKS 1})}
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 1)
+ {TOPOLOGY-TASKS 1})}
{}))
(is (thrown? InvalidTopologyException
(submit-local-topology (:nimbus cluster)
"test/aaa"
{}
topology)))
- (bind topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true)
- :parallelism-hint 16
- :conf {TOPOLOGY-TASKS 16})}
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 16)
+ {TOPOLOGY-TASKS 16})}
{}))
(bind state (:storm-cluster-state cluster))
(is (thrown? InvalidTopologyException
@@ -1042,10 +1109,10 @@
"test"
{TOPOLOGY-WORKERS 3}
topology)))
- (bind topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true)
- :parallelism-hint 5
- :conf {TOPOLOGY-TASKS 5})}
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 5)
+ {TOPOLOGY-TASKS 5})}
{}))
(is (thrown? InvalidTopologyException
(submit-local-topology (:nimbus cluster)
@@ -1080,8 +1147,9 @@
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)}
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 3))}
{}))
(submit-local-topology nimbus "t1" {} topology)
(submit-local-topology nimbus "t2" {} topology)
@@ -1152,8 +1220,9 @@
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)}
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 3))}
{}))
(with-open [_ (MockedZookeeper. (proxy [Zookeeper] []
@@ -1204,7 +1273,7 @@
"org.apache.storm.security.auth.authorizer.DenyAuthorizer"}]
(let [
nimbus (:nimbus cluster)
- topology (thrift/mk-topology {} {})
+ topology (Thrift/buildTopology {} {})
]
(is (thrown? AuthorizationException
(submit-local-topology-with-opts nimbus "mystorm" {} topology
@@ -1220,7 +1289,7 @@
"org.apache.storm.security.auth.authorizer.DenyAuthorizer"}]
(let [
nimbus (:nimbus cluster)
- topology (thrift/mk-topology {} {})
+ topology (Thrift/buildTopology {} {})
]
; Fake good authorization as part of setup.
(mocking [nimbus/check-authorization!]
@@ -1247,7 +1316,7 @@
:daemon-conf {NIMBUS-AUTHORIZER "org.apache.storm.security.auth.authorizer.NoopAuthorizer"}]
(let [nimbus (:nimbus cluster)
topology-name "test-nimbus-check-autho-params"
- topology (thrift/mk-topology {} {})]
+ topology (Thrift/buildTopology {} {})]
(submit-local-topology-with-opts nimbus topology-name {} topology
(SubmitOptions. TopologyInitialStatus/INACTIVE))
@@ -1432,7 +1501,7 @@
(deftest test-validate-topo-config-on-submit
(with-local-cluster [cluster]
(let [nimbus (:nimbus cluster)
- topology (thrift/mk-topology {} {})
+ topology (Thrift/buildTopology {} {})
bad-config {"topology.isolate.machines" "2"}]
; Fake good authorization as part of setup.
(mocking [nimbus/check-authorization!]
@@ -1453,8 +1522,9 @@
(bind cluster-state (cluster/mk-storm-cluster-state conf))
(bind nimbus (nimbus/service-handler conf (nimbus/standalone-nimbus)))
(Time/sleepSecs 1)
- (bind topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3)}
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 3))}
{}))
(submit-local-topology nimbus "t1" {TOPOLOGY-MESSAGE-TIMEOUT-SECS 30} topology)
; make transition for topology t1 to be killed -> nimbus applies this event to cluster state
@@ -1486,8 +1556,9 @@
(bind nimbus (nimbus/service-handler conf (nimbus/standalone-nimbus)))
(bind notifier (InMemoryTopologyActionNotifier.))
(Time/sleepSecs 1)
- (bind topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3)}
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 3))}
{}))
(submit-local-topology nimbus "test-notification" {TOPOLOGY-MESSAGE-TIMEOUT-SECS 30} topology)
@@ -1512,8 +1583,9 @@
(deftest test-debug-on-component
(with-local-cluster [cluster]
(let [nimbus (:nimbus cluster)
- topology (thrift/mk-topology
- {"spout" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3)}
+ topology (Thrift/buildTopology
+ {"spout" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 3))}
{})]
(submit-local-topology nimbus "t1" {TOPOLOGY-WORKERS 1} topology)
(.debug nimbus "t1" "spout" true 100))))
@@ -1521,8 +1593,9 @@
(deftest test-debug-on-global
(with-local-cluster [cluster]
(let [nimbus (:nimbus cluster)
- topology (thrift/mk-topology
- {"spout" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3)}
+ topology (Thrift/buildTopology
+ {"spout" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 3))}
{})]
(submit-local-topology nimbus "t1" {TOPOLOGY-WORKERS 1} topology)
(.debug nimbus "t1" "" true 100))))
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/test/clj/org/apache/storm/scheduler/resource_aware_scheduler_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/org/apache/storm/scheduler/resource_aware_scheduler_test.clj b/storm-core/test/clj/org/apache/storm/scheduler/resource_aware_scheduler_test.clj
index f613a5b..4ca0721 100644
--- a/storm-core/test/clj/org/apache/storm/scheduler/resource_aware_scheduler_test.clj
+++ b/storm-core/test/clj/org/apache/storm/scheduler/resource_aware_scheduler_test.clj
@@ -15,7 +15,8 @@
;; limitations under the License.
(ns org.apache.storm.scheduler.resource-aware-scheduler-test
(:use [clojure test])
- (:use [org.apache.storm util config testing thrift])
+ (:use [org.apache.storm util config testing])
+ (:use [org.apache.storm.internal thrift])
(:require [org.apache.storm.util :refer [map-val]])
(:require [org.apache.storm.daemon [nimbus :as nimbus]])
(:import [org.apache.storm.generated StormTopology]
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/test/clj/org/apache/storm/supervisor_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/org/apache/storm/supervisor_test.clj b/storm-core/test/clj/org/apache/storm/supervisor_test.clj
index 9c31ddf..345cb24 100644
--- a/storm-core/test/clj/org/apache/storm/supervisor_test.clj
+++ b/storm-core/test/clj/org/apache/storm/supervisor_test.clj
@@ -31,10 +31,12 @@
[org.apache.storm.utils.staticmocking ConfigUtilsInstaller
UtilsInstaller])
(:import [java.nio.file.attribute FileAttribute])
+ (:import [org.apache.storm Thrift])
+ (:import [org.apache.storm.utils Utils])
(:use [org.apache.storm config testing util timer log])
(:use [org.apache.storm.daemon common])
(:require [org.apache.storm.daemon [worker :as worker] [supervisor :as supervisor]]
- [org.apache.storm [thrift :as thrift] [cluster :as cluster]])
+ [org.apache.storm [cluster :as cluster]])
(:use [conjure core])
(:require [clojure.java.io :as io]))
@@ -103,8 +105,9 @@
SUPERVISOR-WORKER-TIMEOUT-SECS 15
SUPERVISOR-MONITOR-FREQUENCY-SECS 3}]
(letlocals
- (bind topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 4)}
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 4))}
{}))
(bind sup1 (add-supervisor cluster :id "sup1" :ports [1 2 3 4]))
(bind changed (capture-changed-workers
@@ -156,11 +159,13 @@
SUPERVISOR-WORKER-TIMEOUT-SECS 15
SUPERVISOR-MONITOR-FREQUENCY-SECS 3}]
(letlocals
- (bind topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 4)}
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 4))}
{}))
- (bind topology2 (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 3)}
+ (bind topology2 (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 3))}
{}))
(bind sup1 (add-supervisor cluster :id "sup1" :ports [1 2 3 4]))
(bind sup2 (add-supervisor cluster :id "sup2" :ports [1 2]))
@@ -270,8 +275,9 @@
(check-heartbeat cluster "sup" 3)
(advance-cluster-time cluster 15)
(check-heartbeat cluster "sup" 3)
- (bind topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 4)}
+ (bind topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 4))}
{}))
;; prevent them from launching by capturing them
(capture-changed-workers
@@ -646,7 +652,7 @@
(supervisor/supervisor-data auth-conf nil fake-isupervisor)
(verify-call-times-for cluster/mk-storm-cluster-state 1)
(verify-first-call-args-for-indices cluster/mk-storm-cluster-state [2]
- expected-acls)))))
+ expected-acls))))))
(deftest test-write-log-metadata
(testing "supervisor writes correct data to logs metadata file"
@@ -769,66 +775,68 @@
childopts-with-ids (supervisor/substitute-childopts childopts worker-id topology-id port mem-onheap)]
(is (= expected-childopts childopts-with-ids)))))
- (deftest test-retry-read-assignments
- (with-simulated-time-local-cluster [cluster
- :supervisors 0
- :ports-per-supervisor 2
- :daemon-conf {ConfigUtils/NIMBUS_DO_NOT_REASSIGN true
- NIMBUS-MONITOR-FREQ-SECS 10
- TOPOLOGY-MESSAGE-TIMEOUT-SECS 30
- TOPOLOGY-ACKER-EXECUTORS 0}]
- (letlocals
- (bind sup1 (add-supervisor cluster :id "sup1" :ports [1 2 3 4]))
- (bind topology1 (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 2)}
- {}))
- (bind topology2 (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestPlannerSpout. true) :parallelism-hint 2)}
- {}))
- (bind state (:storm-cluster-state cluster))
- (bind changed (capture-changed-workers
- (submit-mocked-assignment
- (:nimbus cluster)
- (:storm-cluster-state cluster)
- "topology1"
- {TOPOLOGY-WORKERS 2}
- topology1
- {1 "1"
- 2 "1"}
- {[1 1] ["sup1" 1]
- [2 2] ["sup1" 2]}
- {["sup1" 1] [0.0 0.0 0.0]
- ["sup1" 2] [0.0 0.0 0.0]
- })
- (submit-mocked-assignment
- (:nimbus cluster)
- (:storm-cluster-state cluster)
- "topology2"
- {TOPOLOGY-WORKERS 2}
- topology2
- {1 "1"
- 2 "1"}
- {[1 1] ["sup1" 1]
- [2 2] ["sup1" 2]}
- {["sup1" 1] [0.0 0.0 0.0]
- ["sup1" 2] [0.0 0.0 0.0]
- })
- ;; Instead of sleeping until topology is scheduled, rebalance topology so mk-assignments is called.
- (.rebalance (:nimbus cluster) "topology1" (doto (RebalanceOptions.) (.set_wait_secs 0)))
- ))
- (is (empty? (:launched changed)))
- (bind options (RebalanceOptions.))
- (.set_wait_secs options 0)
- (bind changed (capture-changed-workers
- (.rebalance (:nimbus cluster) "topology2" options)
- (advance-cluster-time cluster 10)
- (heartbeat-workers cluster "sup1" [1 2 3 4])
- (advance-cluster-time cluster 10)
- ))
- (validate-launched-once (:launched changed)
- {"sup1" [1 2]}
- (get-storm-id (:storm-cluster-state cluster) "topology1"))
- (validate-launched-once (:launched changed)
- {"sup1" [3 4]}
- (get-storm-id (:storm-cluster-state cluster) "topology2"))
- ))))
+(deftest test-retry-read-assignments
+ (with-simulated-time-local-cluster [cluster
+ :supervisors 0
+ :ports-per-supervisor 2
+ :daemon-conf {ConfigUtils/NIMBUS_DO_NOT_REASSIGN true
+ NIMBUS-MONITOR-FREQ-SECS 10
+ TOPOLOGY-MESSAGE-TIMEOUT-SECS 30
+ TOPOLOGY-ACKER-EXECUTORS 0}]
+ (letlocals
+ (bind sup1 (add-supervisor cluster :id "sup1" :ports [1 2 3 4]))
+ (bind topology1 (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 2))}
+ {}))
+ (bind topology2 (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. true) (Integer. 2))}
+ {}))
+ (bind state (:storm-cluster-state cluster))
+ (bind changed (capture-changed-workers
+ (submit-mocked-assignment
+ (:nimbus cluster)
+ (:storm-cluster-state cluster)
+ "topology1"
+ {TOPOLOGY-WORKERS 2}
+ topology1
+ {1 "1"
+ 2 "1"}
+ {[1 1] ["sup1" 1]
+ [2 2] ["sup1" 2]}
+ {["sup1" 1] [0.0 0.0 0.0]
+ ["sup1" 2] [0.0 0.0 0.0]
+ })
+ (submit-mocked-assignment
+ (:nimbus cluster)
+ (:storm-cluster-state cluster)
+ "topology2"
+ {TOPOLOGY-WORKERS 2}
+ topology2
+ {1 "1"
+ 2 "1"}
+ {[1 1] ["sup1" 1]
+ [2 2] ["sup1" 2]}
+ {["sup1" 1] [0.0 0.0 0.0]
+ ["sup1" 2] [0.0 0.0 0.0]
+ })
+ ;; Instead of sleeping until topology is scheduled, rebalance topology so mk-assignments is called.
+ (.rebalance (:nimbus cluster) "topology1" (doto (RebalanceOptions.) (.set_wait_secs 0)))
+ ))
+ (is (empty? (:launched changed)))
+ (bind options (RebalanceOptions.))
+ (.set_wait_secs options 0)
+ (bind changed (capture-changed-workers
+ (.rebalance (:nimbus cluster) "topology2" options)
+ (advance-cluster-time cluster 10)
+ (heartbeat-workers cluster "sup1" [1 2 3 4])
+ (advance-cluster-time cluster 10)
+ ))
+ (validate-launched-once (:launched changed)
+ {"sup1" [1 2]}
+ (get-storm-id (:storm-cluster-state cluster) "topology1"))
+ (validate-launched-once (:launched changed)
+ {"sup1" [3 4]}
+ (get-storm-id (:storm-cluster-state cluster) "topology2"))
+ )))
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/test/clj/org/apache/storm/tick_tuple_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/org/apache/storm/tick_tuple_test.clj b/storm-core/test/clj/org/apache/storm/tick_tuple_test.clj
index 543db09..99ad957 100644
--- a/storm-core/test/clj/org/apache/storm/tick_tuple_test.clj
+++ b/storm-core/test/clj/org/apache/storm/tick_tuple_test.clj
@@ -15,9 +15,11 @@
;; limitations under the License.
(ns org.apache.storm.tick-tuple-test
(:use [clojure test])
- (:use [org.apache.storm testing clojure config])
+ (:use [org.apache.storm testing config])
+ (:use [org.apache.storm.internal clojure])
(:use [org.apache.storm.daemon common])
- (:require [org.apache.storm [thrift :as thrift]]))
+ (:import [org.apache.storm Thrift])
+ (:import [org.apache.storm.utils Utils]))
(defbolt noop-bolt ["tuple"] {:prepare true}
[conf context collector]
@@ -31,9 +33,12 @@
(deftest test-tick-tuple-works-with-system-bolt
(with-simulated-time-local-cluster [cluster]
- (let [topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec noop-spout)}
- {"2" (thrift/mk-bolt-spec {"1" ["tuple"]} noop-bolt)})]
+ (let [topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails noop-spout)}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareFieldsGrouping ["tuple"])}
+ noop-bolt)})]
(try
(submit-local-topology (:nimbus cluster)
"test"
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/test/clj/org/apache/storm/transactional_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/org/apache/storm/transactional_test.clj b/storm-core/test/clj/org/apache/storm/transactional_test.clj
index dd46a7d..b8af518 100644
--- a/storm-core/test/clj/org/apache/storm/transactional_test.clj
+++ b/storm-core/test/clj/org/apache/storm/transactional_test.clj
@@ -36,7 +36,8 @@
(:import [org.mockito Matchers Mockito])
(:import [org.mockito.exceptions.base MockitoAssertionError])
(:import [java.util HashMap Collections ArrayList])
- (:use [org.apache.storm testing util config clojure])
+ (:use [org.apache.storm testing util config])
+ (:use [org.apache.storm.internal clojure])
(:use [org.apache.storm.daemon common]))
;; Testing TODO:
[2/5] storm git commit: backport thrift.clj to Thrift.java
Posted by bo...@apache.org.
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/clj/org/apache/storm/thrift.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/thrift.clj b/storm-core/src/clj/org/apache/storm/thrift.clj
deleted file mode 100644
index 779c1d1..0000000
--- a/storm-core/src/clj/org/apache/storm/thrift.clj
+++ /dev/null
@@ -1,286 +0,0 @@
-;; Licensed to the Apache Software Foundation (ASF) under one
-;; or more contributor license agreements. See the NOTICE file
-;; distributed with this work for additional information
-;; regarding copyright ownership. The ASF licenses this file
-;; to you under the Apache License, Version 2.0 (the
-;; "License"); you may not use this file except in compliance
-;; with the License. You may obtain a copy of the License at
-;;
-;; http://www.apache.org/licenses/LICENSE-2.0
-;;
-;; Unless required by applicable law or agreed to in writing, software
-;; distributed under the License is distributed on an "AS IS" BASIS,
-;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-;; See the License for the specific language governing permissions and
-;; limitations under the License.
-
-(ns org.apache.storm.thrift
- (:import [java.util HashMap]
- [java.io Serializable]
- [org.apache.storm.generated NodeInfo Assignment])
- (:import [org.apache.storm.generated JavaObject Grouping Nimbus StormTopology
- StormTopology$_Fields Bolt Nimbus$Client Nimbus$Iface
- ComponentCommon Grouping$_Fields SpoutSpec NullStruct StreamInfo
- GlobalStreamId ComponentObject ComponentObject$_Fields
- ShellComponent SupervisorInfo])
- (:import [org.apache.storm.utils Utils NimbusClient ConfigUtils])
- (:import [org.apache.storm Constants])
- (:import [org.apache.storm.security.auth ReqContext])
- (:import [org.apache.storm.grouping CustomStreamGrouping])
- (:import [org.apache.storm.topology TopologyBuilder])
- (:import [org.apache.storm.clojure RichShellBolt RichShellSpout])
- (:import [org.apache.thrift.transport TTransport]
- (org.json.simple JSONValue))
- (:use [org.apache.storm util config log zookeeper]))
-
-(defn instantiate-java-object
- [^JavaObject obj]
- (let [name (symbol (.get_full_class_name obj))
- args (map (memfn getFieldValue) (.get_args_list obj))]
- (eval `(new ~name ~@args))))
-
-(def grouping-constants
- {Grouping$_Fields/FIELDS :fields
- Grouping$_Fields/SHUFFLE :shuffle
- Grouping$_Fields/ALL :all
- Grouping$_Fields/NONE :none
- Grouping$_Fields/CUSTOM_SERIALIZED :custom-serialized
- Grouping$_Fields/CUSTOM_OBJECT :custom-object
- Grouping$_Fields/DIRECT :direct
- Grouping$_Fields/LOCAL_OR_SHUFFLE :local-or-shuffle})
-
-(defn grouping-type
- [^Grouping grouping]
- (grouping-constants (.getSetField grouping)))
-
-(defn field-grouping
- [^Grouping grouping]
- (when-not (= (grouping-type grouping) :fields)
- (throw (IllegalArgumentException. "Tried to get grouping fields from non fields grouping")))
- (.get_fields grouping))
-
-(defn global-grouping?
- [^Grouping grouping]
- (and (= :fields (grouping-type grouping))
- (empty? (field-grouping grouping))))
-
-(defn parallelism-hint
- [^ComponentCommon component-common]
- (let [phint (.get_parallelism_hint component-common)]
- (if-not (.is_set_parallelism_hint component-common) 1 phint)))
-
-(defn nimbus-client-and-conn
- ([host port]
- (nimbus-client-and-conn host port nil))
- ([host port as-user]
- (log-message "Connecting to Nimbus at " host ":" port " as user: " as-user)
- (let [conf (clojurify-structure (ConfigUtils/readStormConfig))
- nimbusClient (NimbusClient. conf host port nil as-user)
- client (.getClient nimbusClient)
- transport (.transport nimbusClient)]
- [client transport] )))
-
-(defmacro with-nimbus-connection
- [[client-sym host port] & body]
- `(let [[^Nimbus$Client ~client-sym ^TTransport conn#] (nimbus-client-and-conn ~host ~port)]
- (try
- ~@body
- (finally (.close conn#)))))
-
-(defmacro with-configured-nimbus-connection
- [client-sym & body]
- `(let [conf# (clojurify-structure (ConfigUtils/readStormConfig))
- context# (ReqContext/context)
- user# (if (.principal context#) (.getName (.principal context#)))
- nimbusClient# (NimbusClient/getConfiguredClientAs conf# user#)
- ~client-sym (.getClient nimbusClient#)
- conn# (.transport nimbusClient#)
- ]
- (try
- ~@body
- (finally (.close conn#)))))
-
-(defn direct-output-fields
- [fields]
- (StreamInfo. fields true))
-
-(defn output-fields
- [fields]
- (StreamInfo. fields false))
-
-;TODO: when translating this function, you should replace the map-val with a proper for loop HERE
-(defn mk-output-spec
- [output-spec]
- (let [output-spec (if (map? output-spec)
- output-spec
- {Utils/DEFAULT_STREAM_ID output-spec})]
- (map-val
- (fn [out]
- (if (instance? StreamInfo out)
- out
- (StreamInfo. out false)))
- output-spec)))
-
-(defnk mk-plain-component-common
- [inputs output-spec parallelism-hint :conf nil]
- (let [ret (ComponentCommon. (HashMap. inputs) (HashMap. (mk-output-spec output-spec)))]
- (when parallelism-hint
- (.set_parallelism_hint ret parallelism-hint))
- (when conf
- (.set_json_conf ret (JSONValue/toJSONString conf)))
- ret))
-
-(defnk mk-spout-spec*
- [spout outputs :p nil :conf nil]
- (SpoutSpec. (ComponentObject/serialized_java (Utils/javaSerialize spout))
- (mk-plain-component-common {} outputs p :conf conf)))
-
-(defn mk-shuffle-grouping
- []
- (Grouping/shuffle (NullStruct.)))
-
-(defn mk-local-or-shuffle-grouping
- []
- (Grouping/local_or_shuffle (NullStruct.)))
-
-(defn mk-fields-grouping
- [fields]
- (Grouping/fields fields))
-
-(defn mk-global-grouping
- []
- (mk-fields-grouping []))
-
-(defn mk-direct-grouping
- []
- (Grouping/direct (NullStruct.)))
-
-(defn mk-all-grouping
- []
- (Grouping/all (NullStruct.)))
-
-(defn mk-none-grouping
- []
- (Grouping/none (NullStruct.)))
-
-(defn deserialized-component-object
- [^ComponentObject obj]
- (when (not= (.getSetField obj) ComponentObject$_Fields/SERIALIZED_JAVA)
- (throw (RuntimeException. "Cannot deserialize non-java-serialized object")))
- (Utils/javaDeserialize (.get_serialized_java obj) Serializable))
-
-(defn serialize-component-object
- [obj]
- (ComponentObject/serialized_java (Utils/javaSerialize obj)))
-
-(defn- mk-grouping
- [grouping-spec]
- (cond (nil? grouping-spec)
- (mk-none-grouping)
-
- (instance? Grouping grouping-spec)
- grouping-spec
-
- (instance? CustomStreamGrouping grouping-spec)
- (Grouping/custom_serialized (Utils/javaSerialize grouping-spec))
-
- (instance? JavaObject grouping-spec)
- (Grouping/custom_object grouping-spec)
-
- (sequential? grouping-spec)
- (mk-fields-grouping grouping-spec)
-
- (= grouping-spec :shuffle)
- (mk-shuffle-grouping)
-
- (= grouping-spec :local-or-shuffle)
- (mk-local-or-shuffle-grouping)
- (= grouping-spec :none)
- (mk-none-grouping)
-
- (= grouping-spec :all)
- (mk-all-grouping)
-
- (= grouping-spec :global)
- (mk-global-grouping)
-
- (= grouping-spec :direct)
- (mk-direct-grouping)
-
- true
- (throw (IllegalArgumentException.
- (str grouping-spec " is not a valid grouping")))))
-
-(defn- mk-inputs
- [inputs]
- (into {} (for [[stream-id grouping-spec] inputs]
- [(if (sequential? stream-id)
- (GlobalStreamId. (first stream-id) (second stream-id))
- (GlobalStreamId. stream-id Utils/DEFAULT_STREAM_ID))
- (mk-grouping grouping-spec)])))
-
-(defnk mk-bolt-spec*
- [inputs bolt outputs :p nil :conf nil]
- (let [common (mk-plain-component-common (mk-inputs inputs) outputs p :conf conf)]
- (Bolt. (ComponentObject/serialized_java (Utils/javaSerialize bolt))
- common)))
-
-(defnk mk-spout-spec
- [spout :parallelism-hint nil :p nil :conf nil]
- (let [parallelism-hint (if p p parallelism-hint)]
- {:obj spout :p parallelism-hint :conf conf}))
-
-(defn- shell-component-params
- [command script-or-output-spec kwargs]
- (if (string? script-or-output-spec)
- [(into-array String [command script-or-output-spec])
- (first kwargs)
- (rest kwargs)]
- [(into-array String command)
- script-or-output-spec
- kwargs]))
-
-(defnk mk-bolt-spec
- [inputs bolt :parallelism-hint nil :p nil :conf nil]
- (let [parallelism-hint (if p p parallelism-hint)]
- {:obj bolt :inputs inputs :p parallelism-hint :conf conf}))
-
-(defn mk-shell-bolt-spec
- [inputs command script-or-output-spec & kwargs]
- (let [[command output-spec kwargs]
- (shell-component-params command script-or-output-spec kwargs)]
- (apply mk-bolt-spec inputs
- (RichShellBolt. command (mk-output-spec output-spec)) kwargs)))
-
-(defn mk-shell-spout-spec
- [command script-or-output-spec & kwargs]
- (let [[command output-spec kwargs]
- (shell-component-params command script-or-output-spec kwargs)]
- (apply mk-spout-spec
- (RichShellSpout. command (mk-output-spec output-spec)) kwargs)))
-
-(defn- add-inputs
- [declarer inputs]
- (doseq [[id grouping] (mk-inputs inputs)]
- (.grouping declarer id grouping)))
-
-(defn mk-topology
- ([spout-map bolt-map]
- (let [builder (TopologyBuilder.)]
- (doseq [[name {spout :obj p :p conf :conf}] spout-map]
- (-> builder (.setSpout name spout (if-not (nil? p) (int p) p)) (.addConfigurations conf)))
- (doseq [[name {bolt :obj p :p conf :conf inputs :inputs}] bolt-map]
- (-> builder (.setBolt name bolt (if-not (nil? p) (int p) p)) (.addConfigurations conf) (add-inputs inputs)))
- (.createTopology builder)))
- ([spout-map bolt-map state-spout-map]
- (mk-topology spout-map bolt-map)))
-
-;; clojurify-structure is needed or else every element becomes the same after successive calls
-;; don't know why this happens
-(def STORM-TOPOLOGY-FIELDS
- (-> StormTopology/metaDataMap clojurify-structure keys))
-
-(def SPOUT-FIELDS
- [StormTopology$_Fields/SPOUTS
- StormTopology$_Fields/STATE_SPOUTS])
-
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/clj/org/apache/storm/ui/core.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/ui/core.clj b/storm-core/src/clj/org/apache/storm/ui/core.clj
index 1bf85d4..5b5acdb 100644
--- a/storm-core/src/clj/org/apache/storm/ui/core.clj
+++ b/storm-core/src/clj/org/apache/storm/ui/core.clj
@@ -49,7 +49,7 @@
(:require [compojure.route :as route]
[compojure.handler :as handler]
[ring.util.response :as resp]
- [org.apache.storm [thrift :as thrift]])
+ [org.apache.storm.internal [thrift :as thrift]])
(:require [metrics.meters :refer [defmeter mark!]])
(:import [org.apache.commons.lang StringEscapeUtils])
(:import [org.apache.logging.log4j Level])
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/jvm/org/apache/storm/Thrift.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/Thrift.java b/storm-core/src/jvm/org/apache/storm/Thrift.java
new file mode 100644
index 0000000..cde822f
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/Thrift.java
@@ -0,0 +1,351 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm;
+
+import org.apache.storm.generated.Bolt;
+import org.apache.storm.generated.JavaObjectArg;
+import org.apache.storm.generated.SpoutSpec;
+import org.apache.storm.generated.StateSpoutSpec;
+import org.apache.storm.generated.StreamInfo;
+
+import java.lang.reflect.Constructor;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.HashMap;
+import java.io.Serializable;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+
+import org.apache.storm.generated.JavaObject;
+import org.apache.storm.generated.Grouping;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.generated.StormTopology._Fields;
+import org.apache.storm.generated.ComponentCommon;
+import org.apache.storm.generated.NullStruct;
+import org.apache.storm.generated.GlobalStreamId;
+import org.apache.storm.generated.ComponentObject;
+
+import org.apache.storm.task.IBolt;
+import org.apache.storm.topology.BoltDeclarer;
+import org.apache.storm.topology.IRichBolt;
+import org.apache.storm.topology.IBasicBolt;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.SpoutDeclarer;
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.storm.utils.Utils;
+import org.apache.storm.grouping.CustomStreamGrouping;
+import org.apache.storm.topology.TopologyBuilder;
+
+public class Thrift {
+ private static Logger LOG = LoggerFactory.getLogger(Thrift.class);
+
+ private static StormTopology._Fields[] STORM_TOPOLOGY_FIELDS = null;
+ private static StormTopology._Fields[] SPOUT_FIELDS =
+ { StormTopology._Fields.SPOUTS, StormTopology._Fields.STATE_SPOUTS };
+
+ static {
+ Set<_Fields> keys = StormTopology.metaDataMap.keySet();
+ keys.toArray(STORM_TOPOLOGY_FIELDS = new StormTopology._Fields[keys.size()]);
+ }
+
+ public static StormTopology._Fields[] getTopologyFields() {
+ return STORM_TOPOLOGY_FIELDS;
+ }
+
+ public static StormTopology._Fields[] getSpoutFields() {
+ return SPOUT_FIELDS;
+ }
+
+ public static class SpoutDetails {
+ private IRichSpout spout;
+ private Integer parallelism;
+ private Map conf;
+
+ public SpoutDetails(IRichSpout spout, Integer parallelism, Map conf) {
+ this.spout = spout;
+ this.parallelism = parallelism;
+ this.conf = conf;
+ }
+
+ public IRichSpout getSpout() {
+ return spout;
+ }
+
+ public Integer getParallelism() {
+ return parallelism;
+ }
+
+ public Map getConf() {
+ return conf;
+ }
+ }
+
+ public static class BoltDetails {
+ private Object bolt;
+ private Map conf;
+ private Integer parallelism;
+ private Map<GlobalStreamId, Grouping> inputs;
+
+ public BoltDetails(Object bolt, Map conf, Integer parallelism,
+ Map<GlobalStreamId, Grouping> inputs) {
+ this.bolt = bolt;
+ this.conf = conf;
+ this.parallelism = parallelism;
+ this.inputs = inputs;
+ }
+
+ public Object getBolt() {
+ return bolt;
+ }
+
+ public Map getConf() {
+ return conf;
+ }
+
+ public Map<GlobalStreamId, Grouping> getInputs() {
+ return inputs;
+ }
+
+ public Integer getParallelism() {
+ return parallelism;
+ }
+ }
+
+ public static StreamInfo directOutputFields(List<String> fields) {
+ return new StreamInfo(fields, true);
+ }
+
+ public static StreamInfo outputFields(List<String> fields) {
+ return new StreamInfo(fields, false);
+ }
+
+ public static Grouping prepareShuffleGrouping() {
+ return Grouping.shuffle(new NullStruct());
+ }
+
+ public static Grouping prepareLocalOrShuffleGrouping() {
+ return Grouping.local_or_shuffle(new NullStruct());
+ }
+
+ public static Grouping prepareFieldsGrouping(List<String> fields) {
+ return Grouping.fields(fields);
+ }
+
+ public static Grouping prepareGlobalGrouping() {
+ return prepareFieldsGrouping(new ArrayList<String>());
+ }
+
+ public static Grouping prepareDirectGrouping() {
+ return Grouping.direct(new NullStruct());
+ }
+
+ public static Grouping prepareAllGrouping() {
+ return Grouping.all(new NullStruct());
+ }
+
+ public static Grouping prepareNoneGrouping() {
+ return Grouping.none(new NullStruct());
+ }
+
+ public static Grouping prepareCustomStreamGrouping(Object obj) {
+ return Grouping.custom_serialized(Utils.javaSerialize(obj));
+ }
+
+ public static Grouping prepareCustomJavaObjectGrouping(JavaObject obj) {
+ return Grouping.custom_object(obj);
+ }
+
+ public static Object instantiateJavaObject(JavaObject obj) {
+
+ List<JavaObjectArg> args = obj.get_args_list();
+ Class[] paraTypes = new Class[args.size()];
+ Object[] paraValues = new Object[args.size()];
+ for (int i = 0; i < args.size(); i++) {
+ JavaObjectArg arg = args.get(i);
+ paraValues[i] = arg.getFieldValue();
+
+ if (arg.getSetField().equals(JavaObjectArg._Fields.INT_ARG)) {
+ paraTypes[i] = Integer.class;
+ } else if (arg.getSetField().equals(JavaObjectArg._Fields.LONG_ARG)) {
+ paraTypes[i] = Long.class;
+ } else if (arg.getSetField().equals(JavaObjectArg._Fields.STRING_ARG)) {
+ paraTypes[i] = String.class;
+ } else if (arg.getSetField().equals(JavaObjectArg._Fields.BOOL_ARG)) {
+ paraTypes[i] = Boolean.class;
+ } else if (arg.getSetField().equals(JavaObjectArg._Fields.BINARY_ARG)) {
+ paraTypes[i] = ByteBuffer.class;
+ } else if (arg.getSetField().equals(JavaObjectArg._Fields.DOUBLE_ARG)) {
+ paraTypes[i] = Double.class;
+ } else {
+ paraTypes[i] = Object.class;
+ }
+ }
+
+ try {
+ Class clazz = Class.forName(obj.get_full_class_name());
+ Constructor cons = clazz.getConstructor(paraTypes);
+ return cons.newInstance(paraValues);
+ } catch (Exception e) {
+ LOG.error("java object instantiation failed", e);
+ }
+
+ return null;
+
+ }
+
+ public static Grouping._Fields groupingType(Grouping grouping) {
+ return grouping.getSetField();
+ }
+
+ public static List<String> fieldGrouping(Grouping grouping) {
+ if (!Grouping._Fields.FIELDS.equals(groupingType(grouping))) {
+ throw new IllegalArgumentException("Tried to get grouping fields from non fields grouping");
+ }
+ return grouping.get_fields();
+ }
+
+ public static boolean isGlobalGrouping(Grouping grouping) {
+ if (Grouping._Fields.FIELDS.equals(groupingType(grouping))) {
+ return fieldGrouping(grouping).isEmpty();
+ }
+
+ return false;
+ }
+
+ public static int getParallelismHint(ComponentCommon componentCommon) {
+ if (!componentCommon.is_set_parallelism_hint()) {
+ return 1;
+ } else {
+ return componentCommon.get_parallelism_hint();
+ }
+ }
+
+ public static ComponentObject serializeComponentObject(Object obj) {
+ return ComponentObject.serialized_java(Utils.javaSerialize(obj));
+ }
+
+ public static Object deserializeComponentObject(ComponentObject obj) {
+ if (obj.getSetField() != ComponentObject._Fields.SERIALIZED_JAVA) {
+ throw new RuntimeException("Cannot deserialize non-java-serialized object");
+ }
+ return Utils.javaDeserialize(obj.get_serialized_java(), Serializable.class);
+ }
+
+ public static ComponentCommon prepareComponentCommon(Map<GlobalStreamId, Grouping> inputs, Map<String,
+ StreamInfo> outputs, Integer parallelismHint) {
+ return prepareComponentCommon(inputs, outputs, parallelismHint, null);
+ }
+
+ public static ComponentCommon prepareComponentCommon(Map<GlobalStreamId, Grouping> inputs, Map<String, StreamInfo> outputs,
+ Integer parallelismHint, Map conf) {
+ Map<GlobalStreamId, Grouping> mappedInputs = new HashMap<>();
+ Map<String, StreamInfo> mappedOutputs = new HashMap<>();
+ if (inputs != null && !inputs.isEmpty()) {
+ mappedInputs.putAll(inputs);
+ }
+ if (outputs !=null && !outputs.isEmpty()) {
+ mappedOutputs.putAll(outputs);
+ }
+ ComponentCommon component = new ComponentCommon(mappedInputs, mappedOutputs);
+ if (parallelismHint != null) {
+ component.set_parallelism_hint(parallelismHint);
+ }
+ if (conf != null) {
+ component.set_json_conf(JSONValue.toJSONString(conf));
+ }
+ return component;
+ }
+
+ public static SpoutSpec prepareSerializedSpoutDetails(IRichSpout spout, Map<String, StreamInfo> outputs) {
+ return new SpoutSpec(ComponentObject.serialized_java
+ (Utils.javaSerialize(spout)), prepareComponentCommon(new HashMap(), outputs, null, null));
+ }
+
+ public static Bolt prepareSerializedBoltDetails(Map<GlobalStreamId, Grouping> inputs, IBolt bolt, Map<String, StreamInfo> outputs,
+ Integer parallelismHint, Map conf) {
+ ComponentCommon common = prepareComponentCommon(inputs, outputs, parallelismHint, conf);
+ return new Bolt(ComponentObject.serialized_java(Utils.javaSerialize(bolt)), common);
+ }
+
+ public static BoltDetails prepareBoltDetails(Map<GlobalStreamId, Grouping> inputs, Object bolt) {
+ return prepareBoltDetails(inputs, bolt, null, null);
+ }
+
+ public static BoltDetails prepareBoltDetails(Map<GlobalStreamId, Grouping> inputs, Object bolt,
+ Integer parallelismHint) {
+ return prepareBoltDetails(inputs, bolt, parallelismHint, null);
+ }
+
+ public static BoltDetails prepareBoltDetails(Map<GlobalStreamId, Grouping> inputs, Object bolt,
+ Integer parallelismHint, Map conf) {
+ BoltDetails details = new BoltDetails(bolt, conf, parallelismHint, inputs);
+ return details;
+ }
+
+ public static SpoutDetails prepareSpoutDetails(IRichSpout spout) {
+ return prepareSpoutDetails(spout, null, null);
+ }
+
+ public static SpoutDetails prepareSpoutDetails(IRichSpout spout, Integer parallelismHint) {
+ return prepareSpoutDetails(spout, parallelismHint, null);
+ }
+
+ public static SpoutDetails prepareSpoutDetails(IRichSpout spout, Integer parallelismHint, Map conf) {
+ SpoutDetails details = new SpoutDetails(spout, parallelismHint, conf);
+ return details;
+ }
+
+ public static StormTopology buildTopology(HashMap<String, SpoutDetails> spoutMap,
+ HashMap<String, BoltDetails> boltMap, HashMap<String, StateSpoutSpec> stateMap) {
+ return buildTopology(spoutMap, boltMap);
+ }
+
+ private static void addInputs(BoltDeclarer declarer, Map<GlobalStreamId, Grouping> inputs) {
+ for(Entry<GlobalStreamId, Grouping> entry : inputs.entrySet()) {
+ declarer.grouping(entry.getKey(), entry.getValue());
+ }
+ }
+
+ public static StormTopology buildTopology(Map<String, SpoutDetails> spoutMap, Map<String, BoltDetails> boltMap) {
+ TopologyBuilder builder = new TopologyBuilder();
+ for (Entry<String, SpoutDetails> entry : spoutMap.entrySet()) {
+ String spoutID = entry.getKey();
+ SpoutDetails spec = entry.getValue();
+ SpoutDeclarer spoutDeclarer = builder.setSpout(spoutID, spec.getSpout(), spec.getParallelism());
+ spoutDeclarer.addConfigurations(spec.getConf());
+ }
+ for (Entry<String, BoltDetails> entry : boltMap.entrySet()) {
+ String spoutID = entry.getKey();
+ BoltDetails spec = entry.getValue();
+ BoltDeclarer boltDeclarer = null;
+ if (spec.bolt instanceof IRichBolt) {
+ boltDeclarer = builder.setBolt(spoutID, (IRichBolt)spec.getBolt(), spec.getParallelism());
+ } else {
+ boltDeclarer = builder.setBolt(spoutID, (IBasicBolt)spec.getBolt(), spec.getParallelism());
+ }
+ boltDeclarer.addConfigurations(spec.getConf());
+ addInputs(boltDeclarer, spec.getInputs());
+ }
+ return builder.createTopology();
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/jvm/org/apache/storm/testing/NGrouping.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/testing/NGrouping.java b/storm-core/src/jvm/org/apache/storm/testing/NGrouping.java
index 45b263d..06853fe 100644
--- a/storm-core/src/jvm/org/apache/storm/testing/NGrouping.java
+++ b/storm-core/src/jvm/org/apache/storm/testing/NGrouping.java
@@ -28,9 +28,7 @@ public class NGrouping implements CustomStreamGrouping {
int _n;
List<Integer> _outTasks;
- public NGrouping(int n) {
- _n = n;
- }
+ public NGrouping(Integer n) {_n = n;}
@Override
public void prepare(WorkerTopologyContext context, GlobalStreamId stream, List<Integer> targetTasks) {
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/jvm/org/apache/storm/testing/PythonShellMetricsBolt.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/testing/PythonShellMetricsBolt.java b/storm-core/src/jvm/org/apache/storm/testing/PythonShellMetricsBolt.java
index eaf7dc8..4beec48 100644
--- a/storm-core/src/jvm/org/apache/storm/testing/PythonShellMetricsBolt.java
+++ b/storm-core/src/jvm/org/apache/storm/testing/PythonShellMetricsBolt.java
@@ -28,18 +28,22 @@ import org.apache.storm.topology.OutputFieldsDeclarer;
public class PythonShellMetricsBolt extends ShellBolt implements IRichBolt {
private static final long serialVersionUID = 1999209252187463355L;
-
- public PythonShellMetricsBolt(String[] command) {
- super(command);
+
+ public PythonShellMetricsBolt(String[] args) {
+ super(args);
}
+ public PythonShellMetricsBolt(String command, String file) {
+ super(command, file);
+ }
+
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
super.prepare(stormConf, context, collector);
-
+
CountShellMetric cMetric = new CountShellMetric();
context.registerMetric("my-custom-shell-metric", cMetric, 5);
}
-
+
public void declareOutputFields(OutputFieldsDeclarer declarer) {
}
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/jvm/org/apache/storm/testing/PythonShellMetricsSpout.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/testing/PythonShellMetricsSpout.java b/storm-core/src/jvm/org/apache/storm/testing/PythonShellMetricsSpout.java
index ed6de14..657baa6 100644
--- a/storm-core/src/jvm/org/apache/storm/testing/PythonShellMetricsSpout.java
+++ b/storm-core/src/jvm/org/apache/storm/testing/PythonShellMetricsSpout.java
@@ -33,11 +33,15 @@ public class PythonShellMetricsSpout extends ShellSpout implements IRichSpout {
public PythonShellMetricsSpout(String[] command) {
super(command);
}
-
+
+ public PythonShellMetricsSpout(String command, String file) {
+ super(command, file);
+ }
+
@Override
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
super.open(conf, context, collector);
-
+
CountShellMetric cMetric = new CountShellMetric();
context.registerMetric("my-custom-shellspout-metric", cMetric, 5);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/jvm/org/apache/storm/utils/Utils.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/utils/Utils.java b/storm-core/src/jvm/org/apache/storm/utils/Utils.java
index 9a849ea..eca9690 100644
--- a/storm-core/src/jvm/org/apache/storm/utils/Utils.java
+++ b/storm-core/src/jvm/org/apache/storm/utils/Utils.java
@@ -52,7 +52,6 @@ import org.apache.thrift.TSerializer;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
-import org.eclipse.jetty.util.log.Log;
import org.json.simple.JSONValue;
import org.json.simple.parser.ParseException;
import org.slf4j.Logger;
@@ -1456,6 +1455,13 @@ public class Utils {
return number & Integer.MAX_VALUE;
}
+ public static GlobalStreamId getGlobalStreamId(String streamId, String componentId) {
+ if (componentId == null) {
+ return new GlobalStreamId(streamId, DEFAULT_STREAM_ID);
+ }
+ return new GlobalStreamId(streamId, componentId);
+ }
+
public static RuntimeException wrapInRuntime(Exception e){
if (e instanceof RuntimeException){
return (RuntimeException)e;
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/test/clj/integration/org/apache/storm/integration_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/integration/org/apache/storm/integration_test.clj b/storm-core/test/clj/integration/org/apache/storm/integration_test.clj
index 5ba6651..6dce7d6 100644
--- a/storm-core/test/clj/integration/org/apache/storm/integration_test.clj
+++ b/storm-core/test/clj/integration/org/apache/storm/integration_test.clj
@@ -15,26 +15,37 @@
;; limitations under the License.
(ns integration.org.apache.storm.integration-test
(:use [clojure test])
- (:import [org.apache.storm Config])
+ (:import [org.apache.storm Config Thrift])
(:import [org.apache.storm.topology TopologyBuilder])
(:import [org.apache.storm.generated InvalidTopologyException SubmitOptions TopologyInitialStatus RebalanceOptions])
(:import [org.apache.storm.testing TestWordCounter TestWordSpout TestGlobalCount
TestAggregatesCounter TestConfBolt AckFailMapTracker AckTracker TestPlannerSpout])
(:import [org.apache.storm.tuple Fields])
- (:use [org.apache.storm testing config clojure])
+ (:use [org.apache.storm testing config util])
+ (:use [org.apache.storm.internal clojure])
(:use [org.apache.storm.daemon common])
- (:require [org.apache.storm [thrift :as thrift]]))
+ (:import [org.apache.storm Thrift])
+ (:import [org.apache.storm.utils Utils]))
(deftest test-basic-topology
(doseq [zmq-on? [true false]]
(with-simulated-time-local-cluster [cluster :supervisors 4
:daemon-conf {STORM-LOCAL-MODE-ZMQ zmq-on?}]
- (let [topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestWordSpout. true) :parallelism-hint 3)}
- {"2" (thrift/mk-bolt-spec {"1" ["word"]} (TestWordCounter.) :parallelism-hint 4)
- "3" (thrift/mk-bolt-spec {"1" :global} (TestGlobalCount.))
- "4" (thrift/mk-bolt-spec {"2" :global} (TestAggregatesCounter.))
- })
+ (let [topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestWordSpout. true) (Integer. 3))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareFieldsGrouping ["word"])}
+ (TestWordCounter.) (Integer. 4))
+ "3" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareGlobalGrouping)}
+ (TestGlobalCount.))
+ "4" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "2" nil)
+ (Thrift/prepareGlobalGrouping)}
+ (TestAggregatesCounter.))})
results (complete-topology cluster
topology
:mock-sources {"1" [["nathan"] ["bob"] ["joey"] ["nathan"]]}
@@ -60,12 +71,14 @@
(deftest test-multi-tasks-per-executor
(with-simulated-time-local-cluster [cluster :supervisors 4]
- (let [topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestWordSpout. true))}
- {"2" (thrift/mk-bolt-spec {"1" :all} emit-task-id
- :parallelism-hint 3
- :conf {TOPOLOGY-TASKS 6})
- })
+ (let [topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails (TestWordSpout. true))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareAllGrouping)}
+ emit-task-id
+ (Integer. 3)
+ {TOPOLOGY-TASKS 6})})
results (complete-topology cluster
topology
:mock-sources {"1" [["a"]]})]
@@ -98,9 +111,11 @@
(let [feeder (feeder-spout ["field1"])
tracker (AckFailMapTracker.)
_ (.setAckFailDelegate feeder tracker)
- topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec feeder)}
- {"2" (thrift/mk-bolt-spec {"1" :global} ack-every-other)})]
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails feeder)}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareGlobalGrouping)} ack-every-other)})]
(submit-local-topology (:nimbus cluster)
"timeout-tester"
{TOPOLOGY-MESSAGE-TIMEOUT-SECS 10}
@@ -117,24 +132,36 @@
)))
(defn mk-validate-topology-1 []
- (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestWordSpout. true) :parallelism-hint 3)}
- {"2" (thrift/mk-bolt-spec {"1" ["word"]} (TestWordCounter.) :parallelism-hint 4)}))
+ (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails (TestWordSpout. true) (Integer. 3))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareFieldsGrouping ["word"])}
+ (TestWordCounter.) (Integer. 4))}))
(defn mk-invalidate-topology-1 []
- (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestWordSpout. true) :parallelism-hint 3)}
- {"2" (thrift/mk-bolt-spec {"3" ["word"]} (TestWordCounter.) :parallelism-hint 4)}))
+ (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails (TestWordSpout. true) (Integer. 3))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "3" nil)
+ (Thrift/prepareFieldsGrouping ["word"])}
+ (TestWordCounter.) (Integer. 4))}))
(defn mk-invalidate-topology-2 []
- (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestWordSpout. true) :parallelism-hint 3)}
- {"2" (thrift/mk-bolt-spec {"1" ["non-exists-field"]} (TestWordCounter.) :parallelism-hint 4)}))
+ (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails (TestWordSpout. true) (Integer. 3))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareFieldsGrouping ["non-exists-field"])}
+ (TestWordCounter.) (Integer. 4))}))
(defn mk-invalidate-topology-3 []
- (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestWordSpout. true) :parallelism-hint 3)}
- {"2" (thrift/mk-bolt-spec {["1" "non-exists-stream"] ["word"]} (TestWordCounter.) :parallelism-hint 4)}))
+ (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails (TestWordSpout. true) (Integer. 3))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" "non-exists-stream")
+ (Thrift/prepareFieldsGrouping ["word"])}
+ (TestWordCounter.) (Integer. 4))}))
(defn try-complete-wc-topology [cluster topology]
(try (do
@@ -164,10 +191,15 @@
(deftest test-system-stream
;; this test works because mocking a spout splits up the tuples evenly among the tasks
(with-simulated-time-local-cluster [cluster]
- (let [topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestWordSpout. true) :p 3)}
- {"2" (thrift/mk-bolt-spec {"1" ["word"] ["1" "__system"] :global} identity-bolt :p 1)
- })
+ (let [topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestWordSpout. true) (Integer. 3))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareFieldsGrouping ["word"])
+ (Utils/getGlobalStreamId "1" "__system")
+ (Thrift/prepareGlobalGrouping)}
+ identity-bolt (Integer. 1))})
results (complete-topology cluster
topology
:mock-sources {"1" [["a"] ["b"] ["c"]]}
@@ -218,20 +250,38 @@
[feeder3 checker3] (ack-tracking-feeder ["num"])
tracked (mk-tracked-topology
cluster
- (topology
- {"1" (spout-spec feeder1)
- "2" (spout-spec feeder2)
- "3" (spout-spec feeder3)}
- {"4" (bolt-spec {"1" :shuffle} (branching-bolt 2))
- "5" (bolt-spec {"2" :shuffle} (branching-bolt 4))
- "6" (bolt-spec {"3" :shuffle} (branching-bolt 1))
- "7" (bolt-spec
- {"4" :shuffle
- "5" :shuffle
- "6" :shuffle}
+ (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails feeder1)
+ "2" (Thrift/prepareSpoutDetails feeder2)
+ "3" (Thrift/prepareSpoutDetails feeder3)}
+ {"4" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareShuffleGrouping)}
+ (branching-bolt 2))
+ "5" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "2" nil)
+ (Thrift/prepareShuffleGrouping)}
+ (branching-bolt 4))
+ "6" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "3" nil)
+ (Thrift/prepareShuffleGrouping)}
+ (branching-bolt 1))
+ "7" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "4" nil)
+ (Thrift/prepareShuffleGrouping)
+ (Utils/getGlobalStreamId "5" nil)
+ (Thrift/prepareShuffleGrouping)
+ (Utils/getGlobalStreamId "6" nil)
+ (Thrift/prepareShuffleGrouping)}
(agg-bolt 3))
- "8" (bolt-spec {"7" :shuffle} (branching-bolt 2))
- "9" (bolt-spec {"8" :shuffle} ack-bolt)}
+ "8" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "7" nil)
+ (Thrift/prepareShuffleGrouping)}
+ (branching-bolt 2))
+ "9" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "8" nil)
+ (Thrift/prepareShuffleGrouping)}
+ ack-bolt)}
))]
(submit-local-topology (:nimbus cluster)
"acking-test1"
@@ -268,13 +318,21 @@
(let [[feeder checker] (ack-tracking-feeder ["num"])
tracked (mk-tracked-topology
cluster
- (topology
- {"1" (spout-spec feeder)}
- {"2" (bolt-spec {"1" :shuffle} identity-bolt)
- "3" (bolt-spec {"1" :shuffle} identity-bolt)
- "4" (bolt-spec
- {"2" :shuffle
- "3" :shuffle}
+ (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails feeder)}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareShuffleGrouping)}
+ identity-bolt)
+ "3" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareShuffleGrouping)}
+ identity-bolt)
+ "4" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "2" nil)
+ (Thrift/prepareShuffleGrouping)
+ (Utils/getGlobalStreamId "3" nil)
+ (Thrift/prepareShuffleGrouping)}
(agg-bolt 4))}))]
(submit-local-topology (:nimbus cluster)
"test-acking2"
@@ -314,10 +372,13 @@
(let [feeder (feeder-spout ["field1"])
tracker (AckFailMapTracker.)
_ (.setAckFailDelegate feeder tracker)
- topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec feeder)
- "2" (thrift/mk-spout-spec open-tracked-spout)}
- {"3" (thrift/mk-bolt-spec {"1" :global} prepare-tracked-bolt)})]
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails feeder)
+ "2" (Thrift/prepareSpoutDetails open-tracked-spout)}
+ {"3" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareGlobalGrouping)}
+ prepare-tracked-bolt)})]
(reset! bolt-prepared? false)
(reset! spout-opened? false)
@@ -343,10 +404,16 @@
(let [[feeder checker] (ack-tracking-feeder ["num"])
tracked (mk-tracked-topology
cluster
- (topology
- {"1" (spout-spec feeder)}
- {"2" (bolt-spec {"1" :shuffle} dup-anchor)
- "3" (bolt-spec {"2" :shuffle} ack-bolt)}))]
+ (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails feeder)}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareShuffleGrouping)}
+ dup-anchor)
+ "3" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "2" nil)
+ (Thrift/prepareShuffleGrouping)}
+ ack-bolt)}))]
(submit-local-topology (:nimbus cluster)
"test"
{}
@@ -362,36 +429,6 @@
(checker 3)
)))
-;; (defspout ConstantSpout ["val"] {:prepare false}
-;; [collector]
-;; (Time/sleep 100)
-;; (emit-spout! collector [1]))
-
-;; (def errored (atom false))
-;; (def restarted (atom false))
-
-;; (defbolt local-error-checker {} [tuple collector]
-;; (when-not @errored
-;; (reset! errored true)
-;; (println "erroring")
-;; (throw (RuntimeException.)))
-;; (when-not @restarted (println "restarted"))
-;; (reset! restarted true))
-
-;; (deftest test-no-halt-local-mode
-;; (with-simulated-time-local-cluster [cluster]
-;; (let [topology (topology
-;; {1 (spout-spec ConstantSpout)}
-;; {2 (bolt-spec {1 :shuffle} local-error-checker)
-;; })]
-;; (submit-local-topology (:nimbus cluster)
-;; "test"
-;; {}
-;; topology)
-;; (while (not @restarted)
-;; (advance-time-ms! 100))
-;; )))
-
(defspout IncSpout ["word"]
[conf context collector]
(let [state (atom 0)]
@@ -416,23 +453,6 @@
)
)))
-;; (deftest test-clojure-spout
-;; (with-local-cluster [cluster]
-;; (let [nimbus (:nimbus cluster)
-;; top (topology
-;; {1 (spout-spec IncSpout)}
-;; {}
-;; )]
-;; (submit-local-topology nimbus
-;; "spout-test"
-;; {TOPOLOGY-DEBUG true
-;; TOPOLOGY-MESSAGE-TIMEOUT-SECS 3}
-;; top)
-;; (Thread/sleep 10000)
-;; (.killTopology nimbus "spout-test")
-;; (Thread/sleep 10000)
-;; )))
-
(deftest test-kryo-decorators-config
(with-simulated-time-local-cluster [cluster
:daemon-conf {TOPOLOGY-SKIP-MISSING-KRYO-REGISTRATIONS true
@@ -513,11 +533,13 @@
(deftest test-hooks
(with-simulated-time-local-cluster [cluster]
- (let [topology (topology {"1" (spout-spec (TestPlannerSpout. (Fields. ["conf"])))
- }
- {"2" (bolt-spec {"1" :shuffle}
- hooks-bolt)
- })
+ (let [topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. (Fields. ["conf"])))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareShuffleGrouping)}
+ hooks-bolt)})
results (complete-topology cluster
topology
:mock-sources {"1" [[1]
@@ -545,9 +567,12 @@
[feeder checker] (ack-tracking-feeder ["num"])
tracked (mk-tracked-topology
cluster
- (topology
- {"1" (spout-spec feeder)}
- {"2" (bolt-spec {"1" :shuffle} report-errors-bolt)}))
+ (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails feeder)}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareShuffleGrouping)}
+ report-errors-bolt)}))
_ (submit-local-topology (:nimbus cluster)
"test-errors"
{TOPOLOGY-ERROR-THROTTLE-INTERVAL-SECS 10
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/test/clj/integration/org/apache/storm/testing4j_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/integration/org/apache/storm/testing4j_test.clj b/storm-core/test/clj/integration/org/apache/storm/testing4j_test.clj
index e86e893..3b1a48b 100644
--- a/storm-core/test/clj/integration/org/apache/storm/testing4j_test.clj
+++ b/storm-core/test/clj/integration/org/apache/storm/testing4j_test.clj
@@ -15,15 +15,19 @@
;; limitations under the License.
(ns integration.org.apache.storm.testing4j-test
(:use [clojure.test])
- (:use [org.apache.storm config clojure testing])
+ (:use [org.apache.storm config testing util])
+ (:use [org.apache.storm.internal clojure])
(:require [integration.org.apache.storm.integration-test :as it])
- (:require [org.apache.storm.thrift :as thrift])
- (:import [org.apache.storm Testing Config ILocalCluster])
+ (:require [org.apache.storm.internal.thrift :as thrift])
+ (:import [org.apache.storm Testing Config ILocalCluster]
+ [org.apache.storm.generated GlobalStreamId])
(:import [org.apache.storm.tuple Values Tuple])
(:import [org.apache.storm.utils Time Utils])
(:import [org.apache.storm.testing MkClusterParam TestJob MockedSources TestWordSpout
TestWordCounter TestGlobalCount TestAggregatesCounter CompleteTopologyParam
- AckFailMapTracker MkTupleParam]))
+ AckFailMapTracker MkTupleParam])
+ (:import [org.apache.storm.utils Utils])
+ (:import [org.apache.storm Thrift]))
(deftest test-with-simulated-time
(is (= false (Time/isSimulating)))
@@ -69,12 +73,20 @@
(Testing/withSimulatedTimeLocalCluster
(reify TestJob
(^void run [this ^ILocalCluster cluster]
- (let [topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestWordSpout. true) :parallelism-hint 3)}
- {"2" (thrift/mk-bolt-spec {"1" ["word"]} (TestWordCounter.) :parallelism-hint 4)
- "3" (thrift/mk-bolt-spec {"1" :global} (TestGlobalCount.))
- "4" (thrift/mk-bolt-spec {"2" :global} (TestAggregatesCounter.))
- })
+ (let [topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails (TestWordSpout. true) (Integer. 3))}
+ {"2" (Thrift/prepareBoltDetails
+ {(GlobalStreamId. "1" Utils/DEFAULT_STREAM_ID)
+ (Thrift/prepareFieldsGrouping ["word"])}
+ (TestWordCounter.) (Integer. 4))
+ "3" (Thrift/prepareBoltDetails
+ {(GlobalStreamId. "1" Utils/DEFAULT_STREAM_ID)
+ (Thrift/prepareGlobalGrouping)}
+ (TestGlobalCount.))
+ "4" (Thrift/prepareBoltDetails
+ {(GlobalStreamId. "2" Utils/DEFAULT_STREAM_ID)
+ (Thrift/prepareGlobalGrouping)}
+ (TestAggregatesCounter.))})
mocked-sources (doto (MockedSources.)
(.addMockData "1" (into-array Values [(Values. (into-array ["nathan"]))
(Values. (into-array ["bob"]))
@@ -106,13 +118,21 @@
(let [[feeder checker] (it/ack-tracking-feeder ["num"])
tracked (Testing/mkTrackedTopology
cluster
- (topology
- {"1" (spout-spec feeder)}
- {"2" (bolt-spec {"1" :shuffle} it/identity-bolt)
- "3" (bolt-spec {"1" :shuffle} it/identity-bolt)
- "4" (bolt-spec
- {"2" :shuffle
- "3" :shuffle}
+ (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails feeder)}
+ {"2" (Thrift/prepareBoltDetails
+ {(GlobalStreamId. "1" Utils/DEFAULT_STREAM_ID)
+ (Thrift/prepareShuffleGrouping)}
+ it/identity-bolt)
+ "3" (Thrift/prepareBoltDetails
+ {(GlobalStreamId. "1" Utils/DEFAULT_STREAM_ID)
+ (Thrift/prepareShuffleGrouping)}
+ it/identity-bolt)
+ "4" (Thrift/prepareBoltDetails
+ {(GlobalStreamId. "2" Utils/DEFAULT_STREAM_ID)
+ (Thrift/prepareShuffleGrouping)
+ (GlobalStreamId. "3" Utils/DEFAULT_STREAM_ID)
+ (Thrift/prepareShuffleGrouping)}
(it/agg-bolt 4))}))]
(.submitTopology cluster
"test-acking2"
@@ -139,9 +159,12 @@
(let [feeder (feeder-spout ["field1"])
tracker (AckFailMapTracker.)
_ (.setAckFailDelegate feeder tracker)
- topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec feeder)}
- {"2" (thrift/mk-bolt-spec {"1" :global} it/ack-every-other)})
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails feeder)}
+ {"2" (Thrift/prepareBoltDetails
+ {(GlobalStreamId. "1" Utils/DEFAULT_STREAM_ID)
+ (Thrift/prepareGlobalGrouping)}
+ it/ack-every-other)})
storm-conf (doto (Config.)
(.put TOPOLOGY-MESSAGE-TIMEOUT-SECS 10))]
(.submitTopology cluster
@@ -170,9 +193,12 @@
(let [feeder (feeder-spout ["field1"])
tracker (AckFailMapTracker.)
_ (.setAckFailDelegate feeder tracker)
- topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec feeder)}
- {"2" (thrift/mk-bolt-spec {"1" :global} it/ack-every-other)})
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails feeder)}
+ {"2" (Thrift/prepareBoltDetails
+ {(GlobalStreamId. "1" Utils/DEFAULT_STREAM_ID)
+ (Thrift/prepareGlobalGrouping)}
+ it/ack-every-other)})
storm-conf (doto (Config.)
(.put TOPOLOGY-MESSAGE-TIMEOUT-SECS 10)
(.put TOPOLOGY-ENABLE-MESSAGE-TIMEOUTS false))]
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/test/clj/org/apache/storm/clojure_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/org/apache/storm/clojure_test.clj b/storm-core/test/clj/org/apache/storm/clojure_test.clj
index ccec825..13fdeb7 100644
--- a/storm-core/test/clj/org/apache/storm/clojure_test.clj
+++ b/storm-core/test/clj/org/apache/storm/clojure_test.clj
@@ -17,10 +17,12 @@
(:use [clojure test])
(:import [org.apache.storm.testing TestWordSpout TestPlannerSpout]
[org.apache.storm.tuple Fields])
- (:use [org.apache.storm testing clojure config])
+ (:use [org.apache.storm testing config])
+ (:use [org.apache.storm.internal clojure])
(:use [org.apache.storm.daemon common])
- (:require [org.apache.storm [thrift :as thrift]]))
-
+ (:require [org.apache.storm.internal [thrift :as thrift]])
+ (:import [org.apache.storm Thrift])
+ (:import [org.apache.storm.utils Utils]))
(defbolt lalala-bolt1 ["word"] [[val :as tuple] collector]
(let [ret (str val "lalala")]
@@ -56,15 +58,21 @@
(deftest test-clojure-bolt
(with-simulated-time-local-cluster [cluster :supervisors 4]
(let [nimbus (:nimbus cluster)
- topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestWordSpout. false))}
- {"2" (thrift/mk-bolt-spec {"1" :shuffle}
- lalala-bolt1)
- "3" (thrift/mk-bolt-spec {"1" :local-or-shuffle}
- lalala-bolt2)
- "4" (thrift/mk-bolt-spec {"1" :shuffle}
- (lalala-bolt3 "_nathan_"))}
- )
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails (TestWordSpout. false))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareShuffleGrouping)}
+ lalala-bolt1)
+ "3" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareLocalOrShuffleGrouping)}
+ lalala-bolt2)
+ "4" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareShuffleGrouping)}
+ (lalala-bolt3 "_nathan_"))}
+ )
results (complete-topology cluster
topology
:mock-sources {"1" [["david"]
@@ -91,11 +99,12 @@
(deftest test-map-emit
(with-simulated-time-local-cluster [cluster :supervisors 4]
- (let [topology (thrift/mk-topology
- {"words" (thrift/mk-spout-spec (TestWordSpout. false))}
- {"out" (thrift/mk-bolt-spec {"words" :shuffle}
- punctuator-bolt)}
- )
+ (let [topology (Thrift/buildTopology
+ {"words" (Thrift/prepareSpoutDetails (TestWordSpout. false))}
+ {"out" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "words" nil)
+ (Thrift/prepareShuffleGrouping)}
+ punctuator-bolt)})
results (complete-topology cluster
topology
:mock-sources {"words" [["foo"] ["bar"]]}
@@ -115,14 +124,19 @@
(deftest test-component-specific-config-clojure
(with-simulated-time-local-cluster [cluster]
- (let [topology (topology {"1" (spout-spec (TestPlannerSpout. (Fields. ["conf"])) :conf {TOPOLOGY-MESSAGE-TIMEOUT-SECS 40})
- }
- {"2" (bolt-spec {"1" :shuffle}
- (conf-query-bolt {"fake.config" 1
- TOPOLOGY-MAX-TASK-PARALLELISM 2
- TOPOLOGY-MAX-SPOUT-PENDING 10})
- :conf {TOPOLOGY-MAX-SPOUT-PENDING 3})
- })
+ (let [topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. (Fields. ["conf"]))
+ nil
+ {TOPOLOGY-MESSAGE-TIMEOUT-SECS 40})}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareShuffleGrouping)}
+ (conf-query-bolt {"fake.config" 1
+ TOPOLOGY-MAX-TASK-PARALLELISM 2
+ TOPOLOGY-MAX-SPOUT-PENDING 10})
+ nil
+ {TOPOLOGY-MAX-SPOUT-PENDING 3})})
results (complete-topology cluster
topology
:topology-name "test123"
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/test/clj/org/apache/storm/cluster_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/org/apache/storm/cluster_test.clj b/storm-core/test/clj/org/apache/storm/cluster_test.clj
index b146cb0..18e3a80 100644
--- a/storm-core/test/clj/org/apache/storm/cluster_test.clj
+++ b/storm-core/test/clj/org/apache/storm/cluster_test.clj
@@ -30,7 +30,8 @@
(:require [conjure.core])
(:use [conjure core])
(:use [clojure test])
- (:use [org.apache.storm cluster config util testing thrift log]))
+ (:use [org.apache.storm cluster config util testing log])
+ (:use [org.apache.storm.internal thrift]))
(defn mk-config [zk-port]
(merge (clojurify-structure (ConfigUtils/readStormConfig))
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/test/clj/org/apache/storm/drpc_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/org/apache/storm/drpc_test.clj b/storm-core/test/clj/org/apache/storm/drpc_test.clj
index 3dcef7a..6024674 100644
--- a/storm-core/test/clj/org/apache/storm/drpc_test.clj
+++ b/storm-core/test/clj/org/apache/storm/drpc_test.clj
@@ -16,7 +16,8 @@
(ns org.apache.storm.drpc-test
(:use [clojure test])
(:import [org.apache.storm.drpc ReturnResults DRPCSpout
- LinearDRPCTopologyBuilder])
+ LinearDRPCTopologyBuilder]
+ [org.apache.storm.utils ConfigUtils Utils])
(:import [org.apache.storm.topology FailedException])
(:import [org.apache.storm.coordination CoordinatedBolt$FinishedCallback])
(:import [org.apache.storm LocalDRPC LocalCluster])
@@ -25,7 +26,9 @@
[org.apache.storm.utils.staticmocking ConfigUtilsInstaller])
(:import [org.apache.storm.generated DRPCExecutionException])
(:import [java.util.concurrent ConcurrentLinkedQueue])
- (:use [org.apache.storm config testing clojure])
+ (:import [org.apache.storm Thrift])
+ (:use [org.apache.storm config testing])
+ (:use [org.apache.storm.internal clojure])
(:use [org.apache.storm.daemon common drpc])
(:use [conjure core]))
@@ -40,12 +43,16 @@
(let [drpc (LocalDRPC.)
spout (DRPCSpout. "test" drpc)
cluster (LocalCluster.)
- topology (topology
- {"1" (spout-spec spout)}
- {"2" (bolt-spec {"1" :shuffle}
- exclamation-bolt)
- "3" (bolt-spec {"2" :shuffle}
- (ReturnResults.))})]
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails spout)}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareShuffleGrouping)}
+ exclamation-bolt)
+ "3" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "2" nil)
+ (Thrift/prepareGlobalGrouping)}
+ (ReturnResults.))})]
(.submitTopology cluster "test" {} topology)
(is (= "aaa!!!" (.execute drpc "test" "aaa")))
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/test/clj/org/apache/storm/grouping_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/org/apache/storm/grouping_test.clj b/storm-core/test/clj/org/apache/storm/grouping_test.clj
index f2a3f4b..61caf68 100644
--- a/storm-core/test/clj/org/apache/storm/grouping_test.clj
+++ b/storm-core/test/clj/org/apache/storm/grouping_test.clj
@@ -18,9 +18,11 @@
(:import [org.apache.storm.testing TestWordCounter TestWordSpout TestGlobalCount TestAggregatesCounter TestWordBytesCounter NGrouping]
[org.apache.storm.generated JavaObject JavaObjectArg])
(:import [org.apache.storm.grouping LoadMapping])
- (:use [org.apache.storm testing clojure log config])
+ (:use [org.apache.storm testing log config])
+ (:use [org.apache.storm.internal clojure])
(:use [org.apache.storm.daemon common executor])
- (:require [org.apache.storm [thrift :as thrift]]))
+ (:import [org.apache.storm Thrift])
+ (:import [org.apache.storm.utils Utils]))
(deftest test-shuffle
(let [shuffle-fn (mk-shuffle-grouper [(int 1) (int 2)] {TOPOLOGY-DISABLE-LOADAWARE-MESSAGING true} nil "comp" "stream")
@@ -77,12 +79,13 @@
(with-simulated-time-local-cluster [cluster :supervisors 4]
(let [spout-phint 4
bolt-phint 6
- topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestWordSpout. true)
- :parallelism-hint spout-phint)}
- {"2" (thrift/mk-bolt-spec {"1" ["word"]}
- (TestWordBytesCounter.)
- :parallelism-hint bolt-phint)
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestWordSpout. true) (Integer. spout-phint))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareFieldsGrouping ["word"])}
+ (TestWordBytesCounter.) (Integer. spout-phint))
})
results (complete-topology
cluster
@@ -101,12 +104,13 @@
(with-simulated-time-local-cluster [cluster :supervisors 4]
(let [spout-phint 4
bolt-phint 6
- topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestWordSpout. true)
- :parallelism-hint spout-phint)}
- {"2" (thrift/mk-bolt-spec {"1" ["word"]}
- (TestWordBytesCounter.)
- :parallelism-hint bolt-phint)
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestWordSpout. true) (Integer. spout-phint))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareFieldsGrouping ["word"])}
+ (TestWordBytesCounter.) (Integer. bolt-phint))
})
results (complete-topology
cluster
@@ -127,15 +131,21 @@
(deftest test-custom-groupings
(with-simulated-time-local-cluster [cluster]
- (let [topology (topology
- {"1" (spout-spec (TestWordSpout. true))}
- {"2" (bolt-spec {"1" (NGrouping. 2)}
- id-bolt
- :p 4)
- "3" (bolt-spec {"1" (JavaObject. "org.apache.storm.testing.NGrouping"
- [(JavaObjectArg/int_arg 3)])}
- id-bolt
- :p 6)
+ (let [topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestWordSpout. true))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareCustomStreamGrouping (NGrouping. (Integer. 2)))}
+ id-bolt
+ (Integer. 4))
+ "3" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareCustomJavaObjectGrouping
+ (JavaObject. "org.apache.storm.testing.NGrouping"
+ [(JavaObjectArg/int_arg 3)]))}
+ id-bolt
+ (Integer. 6))
})
results (complete-topology cluster
topology
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/test/clj/org/apache/storm/messaging/netty_integration_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/org/apache/storm/messaging/netty_integration_test.clj b/storm-core/test/clj/org/apache/storm/messaging/netty_integration_test.clj
index f75a8e3..7fffd34 100644
--- a/storm-core/test/clj/org/apache/storm/messaging/netty_integration_test.clj
+++ b/storm-core/test/clj/org/apache/storm/messaging/netty_integration_test.clj
@@ -15,10 +15,11 @@
;; limitations under the License.
(ns org.apache.storm.messaging.netty-integration-test
(:use [clojure test])
- (:import [org.apache.storm.messaging TransportFactory])
+ (:import [org.apache.storm.messaging TransportFactory]
+ [org.apache.storm Thrift])
(:import [org.apache.storm.testing TestWordSpout TestGlobalCount])
- (:use [org.apache.storm testing config])
- (:require [org.apache.storm [thrift :as thrift]]))
+ (:import [org.apache.storm.utils Utils])
+ (:use [org.apache.storm testing util config]))
(deftest test-integration
(with-simulated-time-local-cluster [cluster :supervisors 4 :supervisor-slot-port-min 6710
@@ -31,10 +32,13 @@
STORM-MESSAGING-NETTY-MAX-SLEEP-MS 5000
STORM-MESSAGING-NETTY-CLIENT-WORKER-THREADS 1
STORM-MESSAGING-NETTY-SERVER-WORKER-THREADS 1}]
- (let [topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestWordSpout. true) :parallelism-hint 4)}
- {"2" (thrift/mk-bolt-spec {"1" :shuffle} (TestGlobalCount.)
- :parallelism-hint 6)})
+ (let [topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestWordSpout. true) (Integer. 4))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareShuffleGrouping)}
+ (TestGlobalCount.) (Integer. 6))})
results (complete-topology cluster
topology
;; important for test that
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/test/clj/org/apache/storm/messaging_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/org/apache/storm/messaging_test.clj b/storm-core/test/clj/org/apache/storm/messaging_test.clj
index e987688..402ea7f 100644
--- a/storm-core/test/clj/org/apache/storm/messaging_test.clj
+++ b/storm-core/test/clj/org/apache/storm/messaging_test.clj
@@ -18,7 +18,8 @@
(:import [org.apache.storm.testing TestWordCounter TestWordSpout TestGlobalCount TestEventLogSpout TestEventOrderCheckBolt])
(:use [org.apache.storm testing config])
(:use [org.apache.storm.daemon common])
- (:require [org.apache.storm [thrift :as thrift]]))
+ (:import [org.apache.storm Thrift])
+ (:import [org.apache.storm.utils Utils]))
(deftest test-local-transport
(doseq [transport-on? [false true]]
@@ -28,10 +29,13 @@
(if transport-on? true false)
STORM-MESSAGING-TRANSPORT
"org.apache.storm.messaging.netty.Context"}]
- (let [topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec (TestWordSpout. true) :parallelism-hint 2)}
- {"2" (thrift/mk-bolt-spec {"1" :shuffle} (TestGlobalCount.)
- :parallelism-hint 6)
+ (let [topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestWordSpout. true) (Integer. 2))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareShuffleGrouping)}
+ (TestGlobalCount.) (Integer. 6))
})
results (complete-topology cluster
topology
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/test/clj/org/apache/storm/metrics_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/org/apache/storm/metrics_test.clj b/storm-core/test/clj/org/apache/storm/metrics_test.clj
index 9f051f6..c186288 100644
--- a/storm-core/test/clj/org/apache/storm/metrics_test.clj
+++ b/storm-core/test/clj/org/apache/storm/metrics_test.clj
@@ -25,10 +25,12 @@
(:import [org.apache.storm.metric.api.rpc CountShellMetric])
(:import [org.apache.storm.utils Utils])
- (:use [org.apache.storm testing clojure config])
+ (:use [org.apache.storm testing config])
+ (:use [org.apache.storm.internal clojure])
(:use [org.apache.storm.daemon common])
(:use [org.apache.storm.metric testing])
- (:require [org.apache.storm [thrift :as thrift]]))
+ (:import [org.apache.storm Thrift])
+ (:import [org.apache.storm.utils Utils]))
(defbolt acking-bolt {} {:prepare true}
[conf context collector]
@@ -105,9 +107,12 @@
"storm.zookeeper.session.timeout" 60000
}]
(let [feeder (feeder-spout ["field1"])
- topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec feeder)}
- {"2" (thrift/mk-bolt-spec {"1" :global} count-acks)})]
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails feeder)}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareGlobalGrouping)}
+ count-acks)})]
(submit-local-topology (:nimbus cluster) "metrics-tester" {} topology)
(.feed feeder ["a"] 1)
@@ -133,9 +138,12 @@
"storm.zookeeper.session.timeout" 60000
}]
(let [feeder (feeder-spout ["field1"])
- topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec feeder)}
- {"2" (thrift/mk-bolt-spec {"1" :all} count-acks :p 1 :conf {TOPOLOGY-TASKS 2})})]
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails feeder)}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareAllGrouping)}
+ count-acks (Integer. 1) {TOPOLOGY-TASKS 2})})]
(submit-local-topology (:nimbus cluster) "metrics-tester-with-multitasks" {} topology)
(.feed feeder ["a"] 1)
@@ -154,10 +162,9 @@
(assert-buckets! "2" "my-custom-metric" [1 0 0 0 0 0 2] cluster))))
(defn mk-shell-bolt-with-metrics-spec
- [inputs command & kwargs]
- (let [command (into-array String command)]
- (apply thrift/mk-bolt-spec inputs
- (PythonShellMetricsBolt. command) kwargs)))
+ [inputs command file]
+ (Thrift/prepareBoltDetails inputs
+ (PythonShellMetricsBolt. command file)))
(deftest test-custom-metric-with-multilang-py
(with-simulated-time-local-cluster
@@ -167,9 +174,12 @@
"storm.zookeeper.session.timeout" 60000
}]
(let [feeder (feeder-spout ["field1"])
- topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec feeder)}
- {"2" (mk-shell-bolt-with-metrics-spec {"1" :global} ["python" "tester_bolt_metrics.py"])})]
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails feeder)}
+ {"2" (mk-shell-bolt-with-metrics-spec
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareGlobalGrouping)}
+ "python" "tester_bolt_metrics.py")})]
(submit-local-topology (:nimbus cluster) "shell-metrics-tester" {} topology)
(.feed feeder ["a"] 1)
@@ -189,9 +199,8 @@
)))
(defn mk-shell-spout-with-metrics-spec
- [command & kwargs]
- (let [command (into-array String command)]
- (apply thrift/mk-spout-spec (PythonShellMetricsSpout. command) kwargs)))
+ [command file]
+ (Thrift/prepareSpoutDetails (PythonShellMetricsSpout. command file)))
(deftest test-custom-metric-with-spout-multilang-py
(with-simulated-time-local-cluster
@@ -199,9 +208,12 @@
[{"class" "clojure.storm.metric.testing.FakeMetricConsumer"}]
"storm.zookeeper.connection.timeout" 30000
"storm.zookeeper.session.timeout" 60000}]
- (let [topology (thrift/mk-topology
- {"1" (mk-shell-spout-with-metrics-spec ["python" "tester_spout_metrics.py"])}
- {"2" (thrift/mk-bolt-spec {"1" :all} count-acks)})]
+ (let [topology (Thrift/buildTopology
+ {"1" (mk-shell-spout-with-metrics-spec "python" "tester_spout_metrics.py")}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareAllGrouping)}
+ count-acks)})]
(submit-local-topology (:nimbus cluster) "shell-spout-metrics-tester" {} topology)
(advance-cluster-time cluster 7)
@@ -216,9 +228,12 @@
TOPOLOGY-STATS-SAMPLE-RATE 1.0
TOPOLOGY-BUILTIN-METRICS-BUCKET-SIZE-SECS 60}]
(let [feeder (feeder-spout ["field1"])
- topology (thrift/mk-topology
- {"myspout" (thrift/mk-spout-spec feeder)}
- {"mybolt" (thrift/mk-bolt-spec {"myspout" :shuffle} acking-bolt)})]
+ topology (Thrift/buildTopology
+ {"myspout" (Thrift/prepareSpoutDetails feeder)}
+ {"mybolt" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "myspout" nil)
+ (Thrift/prepareShuffleGrouping)}
+ acking-bolt)})]
(submit-local-topology (:nimbus cluster) "metrics-tester" {} topology)
(.feed feeder ["a"] 1)
@@ -255,9 +270,12 @@
(let [feeder (feeder-spout ["field1"])
tracker (AckFailMapTracker.)
_ (.setAckFailDelegate feeder tracker)
- topology (thrift/mk-topology
- {"myspout" (thrift/mk-spout-spec feeder)}
- {"mybolt" (thrift/mk-bolt-spec {"myspout" :shuffle} ack-every-other)})]
+ topology (Thrift/buildTopology
+ {"myspout" (Thrift/prepareSpoutDetails feeder)}
+ {"mybolt" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "myspout" nil)
+ (Thrift/prepareShuffleGrouping)}
+ ack-every-other)})]
(submit-local-topology (:nimbus cluster)
"metrics-tester"
{}
@@ -307,9 +325,12 @@
(let [feeder (feeder-spout ["field1"])
tracker (AckFailMapTracker.)
_ (.setAckFailDelegate feeder tracker)
- topology (thrift/mk-topology
- {"myspout" (thrift/mk-spout-spec feeder)}
- {"mybolt" (thrift/mk-bolt-spec {"myspout" :global} ack-every-other)})]
+ topology (Thrift/buildTopology
+ {"myspout" (Thrift/prepareSpoutDetails feeder)}
+ {"mybolt" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "myspout" nil)
+ (Thrift/prepareGlobalGrouping)}
+ ack-every-other)})]
(submit-local-topology (:nimbus cluster)
"timeout-tester"
{TOPOLOGY-MESSAGE-TIMEOUT-SECS 10}
@@ -341,8 +362,8 @@
[{"class" "clojure.storm.metric.testing.FakeMetricConsumer"}]
TOPOLOGY-BUILTIN-METRICS-BUCKET-SIZE-SECS 60}]
(let [feeder (feeder-spout ["field1"])
- topology (thrift/mk-topology
- {"1" (thrift/mk-spout-spec feeder)}
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails feeder)}
{})]
(submit-local-topology (:nimbus cluster) "metrics-tester" {} topology)
[5/5] storm git commit: Added STORM-1258 to Changelog
Posted by bo...@apache.org.
Added STORM-1258 to Changelog
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/d187a20e
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/d187a20e
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/d187a20e
Branch: refs/heads/master
Commit: d187a20e22983fc8b4433883e895727ce0467ec0
Parents: 5e4c0a7
Author: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Authored: Thu Feb 18 10:00:53 2016 -0600
Committer: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Committed: Thu Feb 18 10:00:53 2016 -0600
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/d187a20e/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 25ecb77..196ed05 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,5 @@
## 2.0.0
+ * STORM-1258: port thrift.clj to Thrift.java
* STORM-1336: Evalute/Port JStorm cgroup support and implement cgroup support for RAS
* STORM-1511: min/max operators implementation in Trident streams API.
* STROM-1263: port backtype.storm.command.kill-topology to java
[3/5] storm git commit: backport thrift.clj to Thrift.java
Posted by bo...@apache.org.
backport thrift.clj to Thrift.java
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/de9cb106
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/de9cb106
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/de9cb106
Branch: refs/heads/master
Commit: de9cb106f3a68f8b13d16a39b242cecd7e5b0513
Parents: 4699990
Author: Sanket <sc...@untilservice-lm>
Authored: Wed Feb 17 10:20:17 2016 -0600
Committer: Sanket <sc...@untilservice-lm>
Committed: Wed Feb 17 10:20:17 2016 -0600
----------------------------------------------------------------------
examples/storm-starter/pom.xml | 10 +
.../org/apache/storm/starter/clj/word_count.clj | 3 +-
pom.xml | 1 +
storm-clojure/pom.xml | 74 ++++
.../src/clj/org/apache/storm/clojure.clj | 207 +++++++++++
.../src/clj/org/apache/storm/thrift.clj | 286 +++++++++++++++
storm-clojure/src/test/clj/clojure_test.clj | 158 +++++++++
storm-core/src/clj/org/apache/storm/clojure.clj | 207 -----------
.../clj/org/apache/storm/command/get_errors.clj | 3 +-
.../clj/org/apache/storm/command/monitor.clj | 2 +-
.../clj/org/apache/storm/command/rebalance.clj | 3 +-
.../org/apache/storm/command/set_log_level.clj | 3 +-
.../apache/storm/command/shell_submission.clj | 2 +-
.../src/clj/org/apache/storm/daemon/common.clj | 121 ++++---
.../clj/org/apache/storm/daemon/executor.clj | 31 +-
.../src/clj/org/apache/storm/daemon/task.clj | 4 +-
.../clj/org/apache/storm/internal/clojure.clj | 201 +++++++++++
.../clj/org/apache/storm/internal/thrift.clj | 96 +++++
storm-core/src/clj/org/apache/storm/testing.clj | 29 +-
storm-core/src/clj/org/apache/storm/thrift.clj | 286 ---------------
storm-core/src/clj/org/apache/storm/ui/core.clj | 2 +-
storm-core/src/jvm/org/apache/storm/Thrift.java | 351 +++++++++++++++++++
.../jvm/org/apache/storm/testing/NGrouping.java | 4 +-
.../storm/testing/PythonShellMetricsBolt.java | 14 +-
.../storm/testing/PythonShellMetricsSpout.java | 8 +-
.../src/jvm/org/apache/storm/utils/Utils.java | 8 +-
.../org/apache/storm/integration_test.clj | 259 +++++++-------
.../org/apache/storm/testing4j_test.clj | 72 ++--
.../test/clj/org/apache/storm/clojure_test.clj | 64 ++--
.../test/clj/org/apache/storm/cluster_test.clj | 3 +-
.../test/clj/org/apache/storm/drpc_test.clj | 23 +-
.../test/clj/org/apache/storm/grouping_test.clj | 56 +--
.../storm/messaging/netty_integration_test.clj | 18 +-
.../clj/org/apache/storm/messaging_test.clj | 14 +-
.../test/clj/org/apache/storm/metrics_test.clj | 85 +++--
.../test/clj/org/apache/storm/nimbus_test.clj | 257 +++++++++-----
.../scheduler/resource_aware_scheduler_test.clj | 3 +-
.../clj/org/apache/storm/supervisor_test.clj | 154 ++++----
.../clj/org/apache/storm/tick_tuple_test.clj | 15 +-
.../clj/org/apache/storm/transactional_test.clj | 3 +-
40 files changed, 2128 insertions(+), 1012 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/examples/storm-starter/pom.xml
----------------------------------------------------------------------
diff --git a/examples/storm-starter/pom.xml b/examples/storm-starter/pom.xml
index 1a7644a..929c8ea 100644
--- a/examples/storm-starter/pom.xml
+++ b/examples/storm-starter/pom.xml
@@ -83,6 +83,16 @@
<version>3.0.3</version>
</dependency>
<dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-clojure</artifactId>
+ <version>${project.version}</version>
+ <!--
+ Use "provided" scope to keep storm out of the jar-with-dependencies
+ For IntelliJ dev, intellij will load properly.
+ -->
+ <scope>${provided.scope}</scope>
+ </dependency>
+ <dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>${project.version}</version>
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/examples/storm-starter/src/clj/org/apache/storm/starter/clj/word_count.clj
----------------------------------------------------------------------
diff --git a/examples/storm-starter/src/clj/org/apache/storm/starter/clj/word_count.clj b/examples/storm-starter/src/clj/org/apache/storm/starter/clj/word_count.clj
index fb3a695..c35cc1f 100644
--- a/examples/storm-starter/src/clj/org/apache/storm/starter/clj/word_count.clj
+++ b/examples/storm-starter/src/clj/org/apache/storm/starter/clj/word_count.clj
@@ -14,7 +14,8 @@
;; See the License for the specific language governing permissions and
;; limitations under the License.
(ns org.apache.storm.starter.clj.word-count
- (:import [org.apache.storm StormSubmitter LocalCluster])
+ (:import [org.apache.storm StormSubmitter LocalCluster]
+ [org.apache.storm.utils Utils])
(:use [org.apache.storm clojure config])
(:gen-class))
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 61a1ed9..6d1a93e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -271,6 +271,7 @@
<module>external/storm-cassandra</module>
<module>external/storm-mqtt</module>
<module>examples/storm-starter</module>
+ <module>storm-clojure</module>
</modules>
<dependencies>
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-clojure/pom.xml
----------------------------------------------------------------------
diff --git a/storm-clojure/pom.xml b/storm-clojure/pom.xml
new file mode 100644
index 0000000..7ce4943
--- /dev/null
+++ b/storm-clojure/pom.xml
@@ -0,0 +1,74 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <artifactId>storm</artifactId>
+ <groupId>org.apache.storm</groupId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>storm-clojure</artifactId>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-core</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.googlecode.json-simple</groupId>
+ <artifactId>json-simple</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>com.theoryinpractise</groupId>
+ <artifactId>clojure-maven-plugin</artifactId>
+ <extensions>true</extensions>
+ <configuration>
+ <sourceDirectories>
+ <sourceDirectory>src/clj</sourceDirectory>
+ </sourceDirectories>
+ </configuration>
+ <executions>
+ <execution>
+ <id>compile</id>
+ <phase>compile</phase>
+ <goals>
+ <goal>compile</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-clojure/src/clj/org/apache/storm/clojure.clj
----------------------------------------------------------------------
diff --git a/storm-clojure/src/clj/org/apache/storm/clojure.clj b/storm-clojure/src/clj/org/apache/storm/clojure.clj
new file mode 100644
index 0000000..9e1836f
--- /dev/null
+++ b/storm-clojure/src/clj/org/apache/storm/clojure.clj
@@ -0,0 +1,207 @@
+;; Licensed to the Apache Software Foundation (ASF) under one
+;; or more contributor license agreements. See the NOTICE file
+;; distributed with this work for additional information
+;; regarding copyright ownership. The ASF licenses this file
+;; to you under the Apache License, Version 2.0 (the
+;; "License"); you may not use this file except in compliance
+;; with the License. You may obtain a copy of the License at
+;;
+;; http://www.apache.org/licenses/LICENSE-2.0
+;;
+;; Unless required by applicable law or agreed to in writing, software
+;; distributed under the License is distributed on an "AS IS" BASIS,
+;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+;; See the License for the specific language governing permissions and
+;; limitations under the License.
+
+(ns org.apache.storm.clojure
+ (:use [org.apache.storm util])
+ (:import [org.apache.storm StormSubmitter])
+ (:import [org.apache.storm.generated StreamInfo])
+ (:import [org.apache.storm.tuple Tuple])
+ (:import [org.apache.storm.task OutputCollector IBolt TopologyContext])
+ (:import [org.apache.storm.spout SpoutOutputCollector ISpout])
+ (:import [org.apache.storm.utils Utils])
+ (:import [org.apache.storm.clojure ClojureBolt ClojureSpout])
+ (:import [java.util Collection List])
+ (:require [org.apache.storm [thrift :as thrift]]))
+
+(defn direct-stream [fields]
+ (StreamInfo. fields true))
+
+(defn to-spec [avar]
+ (let [m (meta avar)]
+ [(str (:ns m)) (str (:name m))]))
+
+(defn clojure-bolt* [output-spec fn-var conf-fn-var args]
+ (ClojureBolt. (to-spec fn-var) (to-spec conf-fn-var) args (thrift/mk-output-spec output-spec)))
+
+(defmacro clojure-bolt [output-spec fn-sym conf-fn-sym args]
+ `(clojure-bolt* ~output-spec (var ~fn-sym) (var ~conf-fn-sym) ~args))
+
+(defn clojure-spout* [output-spec fn-var conf-var args]
+ (let [m (meta fn-var)]
+ (ClojureSpout. (to-spec fn-var) (to-spec conf-var) args (thrift/mk-output-spec output-spec))
+ ))
+
+(defmacro clojure-spout [output-spec fn-sym conf-sym args]
+ `(clojure-spout* ~output-spec (var ~fn-sym) (var ~conf-sym) ~args))
+
+(defn normalize-fns [body]
+ (for [[name args & impl] body
+ :let [args (-> "this"
+ gensym
+ (cons args)
+ vec)]]
+ (concat [name args] impl)
+ ))
+
+(defmacro bolt [& body]
+ (let [[bolt-fns other-fns] (split-with #(not (symbol? %)) body)
+ fns (normalize-fns bolt-fns)]
+ `(reify IBolt
+ ~@fns
+ ~@other-fns)))
+
+(defmacro bolt-execute [& body]
+ `(bolt
+ (~'execute ~@body)))
+
+(defmacro spout [& body]
+ (let [[spout-fns other-fns] (split-with #(not (symbol? %)) body)
+ fns (normalize-fns spout-fns)]
+ `(reify ISpout
+ ~@fns
+ ~@other-fns)))
+
+(defmacro defbolt [name output-spec & [opts & impl :as all]]
+ (if-not (map? opts)
+ `(defbolt ~name ~output-spec {} ~@all)
+ (let [worker-name (symbol (str name "__"))
+ conf-fn-name (symbol (str name "__conf__"))
+ params (:params opts)
+ conf-code (:conf opts)
+ fn-body (if (:prepare opts)
+ (cons 'fn impl)
+ (let [[args & impl-body] impl
+ coll-sym (nth args 1)
+ args (vec (take 1 args))
+ prepargs [(gensym "conf") (gensym "context") coll-sym]]
+ `(fn ~prepargs (bolt (~'execute ~args ~@impl-body)))))
+ definer (if params
+ `(defn ~name [& args#]
+ (clojure-bolt ~output-spec ~worker-name ~conf-fn-name args#))
+ `(def ~name
+ (clojure-bolt ~output-spec ~worker-name ~conf-fn-name []))
+ )
+ ]
+ `(do
+ (defn ~conf-fn-name ~(if params params [])
+ ~conf-code
+ )
+ (defn ~worker-name ~(if params params [])
+ ~fn-body
+ )
+ ~definer
+ ))))
+
+(defmacro defspout [name output-spec & [opts & impl :as all]]
+ (if-not (map? opts)
+ `(defspout ~name ~output-spec {} ~@all)
+ (let [worker-name (symbol (str name "__"))
+ conf-fn-name (symbol (str name "__conf__"))
+ params (:params opts)
+ conf-code (:conf opts)
+ prepare? (:prepare opts)
+ prepare? (if (nil? prepare?) true prepare?)
+ fn-body (if prepare?
+ (cons 'fn impl)
+ (let [[args & impl-body] impl
+ coll-sym (first args)
+ prepargs [(gensym "conf") (gensym "context") coll-sym]]
+ `(fn ~prepargs (spout (~'nextTuple [] ~@impl-body)))))
+ definer (if params
+ `(defn ~name [& args#]
+ (clojure-spout ~output-spec ~worker-name ~conf-fn-name args#))
+ `(def ~name
+ (clojure-spout ~output-spec ~worker-name ~conf-fn-name []))
+ )
+ ]
+ `(do
+ (defn ~conf-fn-name ~(if params params [])
+ ~conf-code
+ )
+ (defn ~worker-name ~(if params params [])
+ ~fn-body
+ )
+ ~definer
+ ))))
+
+(defprotocol TupleValues
+ (tuple-values [values collector stream]))
+
+(extend-protocol TupleValues
+ java.util.Map
+ (tuple-values [this collector ^String stream]
+ (let [^TopologyContext context (:context collector)
+ fields (.. context (getThisOutputFields stream) toList) ]
+ (vec (map (into
+ (empty this) (for [[k v] this]
+ [(if (keyword? k) (name k) k) v]))
+ fields))))
+ java.util.List
+ (tuple-values [this collector stream]
+ this))
+
+(defn- collectify
+ [obj]
+ (if (or (sequential? obj) (instance? Collection obj))
+ obj
+ [obj]))
+
+(defnk emit-bolt! [collector values
+ :stream Utils/DEFAULT_STREAM_ID :anchor []]
+ (let [^List anchor (collectify anchor)
+ values (tuple-values values collector stream) ]
+ (.emit ^OutputCollector (:output-collector collector) stream anchor values)
+ ))
+
+(defnk emit-direct-bolt! [collector task values
+ :stream Utils/DEFAULT_STREAM_ID :anchor []]
+ (let [^List anchor (collectify anchor)
+ values (tuple-values values collector stream) ]
+ (.emitDirect ^OutputCollector (:output-collector collector) task stream anchor values)
+ ))
+
+(defn ack! [collector ^Tuple tuple]
+ (.ack ^OutputCollector (:output-collector collector) tuple))
+
+(defn fail! [collector ^Tuple tuple]
+ (.fail ^OutputCollector (:output-collector collector) tuple))
+
+(defn report-error! [collector ^Tuple tuple]
+ (.reportError ^OutputCollector (:output-collector collector) tuple))
+
+(defnk emit-spout! [collector values
+ :stream Utils/DEFAULT_STREAM_ID :id nil]
+ (let [values (tuple-values values collector stream)]
+ (.emit ^SpoutOutputCollector (:output-collector collector) stream values id)))
+
+(defnk emit-direct-spout! [collector task values
+ :stream Utils/DEFAULT_STREAM_ID :id nil]
+ (let [values (tuple-values values collector stream)]
+ (.emitDirect ^SpoutOutputCollector (:output-collector collector) task stream values id)))
+
+(defalias topology thrift/mk-topology)
+(defalias bolt-spec thrift/mk-bolt-spec)
+(defalias spout-spec thrift/mk-spout-spec)
+(defalias shell-bolt-spec thrift/mk-shell-bolt-spec)
+(defalias shell-spout-spec thrift/mk-shell-spout-spec)
+
+(defn submit-remote-topology [name conf topology]
+ (StormSubmitter/submitTopology name conf topology))
+
+(defn local-cluster []
+ ;; do this to avoid a cyclic dependency of
+ ;; LocalCluster -> testing -> nimbus -> bootstrap -> clojure -> LocalCluster
+ (eval '(new org.apache.storm.LocalCluster)))
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-clojure/src/clj/org/apache/storm/thrift.clj
----------------------------------------------------------------------
diff --git a/storm-clojure/src/clj/org/apache/storm/thrift.clj b/storm-clojure/src/clj/org/apache/storm/thrift.clj
new file mode 100644
index 0000000..bf13d23
--- /dev/null
+++ b/storm-clojure/src/clj/org/apache/storm/thrift.clj
@@ -0,0 +1,286 @@
+;; Licensed to the Apache Software Foundation (ASF) under one
+;; or more contributor license agreements. See the NOTICE file
+;; distributed with this work for additional information
+;; regarding copyright ownership. The ASF licenses this file
+;; to you under the Apache License, Version 2.0 (the
+;; "License"); you may not use this file except in compliance
+;; with the License. You may obtain a copy of the License at
+;;
+;; http://www.apache.org/licenses/LICENSE-2.0
+;;
+;; Unless required by applicable law or agreed to in writing, software
+;; distributed under the License is distributed on an "AS IS" BASIS,
+;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+;; See the License for the specific language governing permissions and
+;; limitations under the License.
+
+(ns org.apache.storm.thrift
+ (:import [java.util HashMap]
+ [java.io Serializable]
+ [org.apache.storm.generated NodeInfo Assignment])
+ (:import [org.apache.storm.generated JavaObject Grouping Nimbus StormTopology
+ StormTopology$_Fields Bolt Nimbus$Client Nimbus$Iface
+ ComponentCommon Grouping$_Fields SpoutSpec NullStruct StreamInfo
+ GlobalStreamId ComponentObject ComponentObject$_Fields
+ ShellComponent SupervisorInfo])
+ (:import [org.apache.storm.utils Utils NimbusClient ConfigUtils])
+ (:import [org.apache.storm Constants])
+ (:import [org.apache.storm.security.auth ReqContext])
+ (:import [org.apache.storm.grouping CustomStreamGrouping])
+ (:import [org.apache.storm.topology TopologyBuilder])
+ (:import [org.apache.storm.clojure RichShellBolt RichShellSpout])
+ (:import [org.apache.storm.thrift.transport TTransport]
+ (org.json.simple JSONValue))
+ (:use [org.apache.storm util config log zookeeper]))
+
+(defn instantiate-java-object
+ [^JavaObject obj]
+ (let [name (symbol (.get_full_class_name obj))
+ args (map (memfn getFieldValue) (.get_args_list obj))]
+ (eval `(new ~name ~@args))))
+
+(def grouping-constants
+ {Grouping$_Fields/FIELDS :fields
+ Grouping$_Fields/SHUFFLE :shuffle
+ Grouping$_Fields/ALL :all
+ Grouping$_Fields/NONE :none
+ Grouping$_Fields/CUSTOM_SERIALIZED :custom-serialized
+ Grouping$_Fields/CUSTOM_OBJECT :custom-object
+ Grouping$_Fields/DIRECT :direct
+ Grouping$_Fields/LOCAL_OR_SHUFFLE :local-or-shuffle})
+
+(defn grouping-type
+ [^Grouping grouping]
+ (grouping-constants (.getSetField grouping)))
+
+(defn field-grouping
+ [^Grouping grouping]
+ (when-not (= (grouping-type grouping) :fields)
+ (throw (IllegalArgumentException. "Tried to get grouping fields from non fields grouping")))
+ (.get_fields grouping))
+
+(defn global-grouping?
+ [^Grouping grouping]
+ (and (= :fields (grouping-type grouping))
+ (empty? (field-grouping grouping))))
+
+(defn parallelism-hint
+ [^ComponentCommon component-common]
+ (let [phint (.get_parallelism_hint component-common)]
+ (if-not (.is_set_parallelism_hint component-common) 1 phint)))
+
+(defn nimbus-client-and-conn
+ ([host port]
+ (nimbus-client-and-conn host port nil))
+ ([host port as-user]
+ (log-message "Connecting to Nimbus at " host ":" port " as user: " as-user)
+ (let [conf (clojurify-structure (ConfigUtils/readStormConfig))
+ nimbusClient (NimbusClient. conf host port nil as-user)
+ client (.getClient nimbusClient)
+ transport (.transport nimbusClient)]
+ [client transport] )))
+
+(defmacro with-nimbus-connection
+ [[client-sym host port] & body]
+ `(let [[^Nimbus$Client ~client-sym ^TTransport conn#] (nimbus-client-and-conn ~host ~port)]
+ (try
+ ~@body
+ (finally (.close conn#)))))
+
+(defmacro with-configured-nimbus-connection
+ [client-sym & body]
+ `(let [conf# (clojurify-structure (ConfigUtils/readStormConfig))
+ context# (ReqContext/context)
+ user# (if (.principal context#) (.getName (.principal context#)))
+ nimbusClient# (NimbusClient/getConfiguredClientAs conf# user#)
+ ~client-sym (.getClient nimbusClient#)
+ conn# (.transport nimbusClient#)
+ ]
+ (try
+ ~@body
+ (finally (.close conn#)))))
+
+(defn direct-output-fields
+ [fields]
+ (StreamInfo. fields true))
+
+(defn output-fields
+ [fields]
+ (StreamInfo. fields false))
+
+;TODO: when translating this function, you should replace the map-val with a proper for loop HERE
+(defn mk-output-spec
+ [output-spec]
+ (let [output-spec (if (map? output-spec)
+ output-spec
+ {Utils/DEFAULT_STREAM_ID output-spec})]
+ (map-val
+ (fn [out]
+ (if (instance? StreamInfo out)
+ out
+ (StreamInfo. out false)))
+ output-spec)))
+
+(defnk mk-plain-component-common
+ [inputs output-spec parallelism-hint :conf nil]
+ (let [ret (ComponentCommon. (HashMap. inputs) (HashMap. (mk-output-spec output-spec)))]
+ (when parallelism-hint
+ (.set_parallelism_hint ret parallelism-hint))
+ (when conf
+ (.set_json_conf ret (JSONValue/toJSONString conf)))
+ ret))
+
+(defnk mk-spout-spec*
+ [spout outputs :p nil :conf nil]
+ (SpoutSpec. (ComponentObject/serialized_java (Utils/javaSerialize spout))
+ (mk-plain-component-common {} outputs p :conf conf)))
+
+(defn mk-shuffle-grouping
+ []
+ (Grouping/shuffle (NullStruct.)))
+
+(defn mk-local-or-shuffle-grouping
+ []
+ (Grouping/local_or_shuffle (NullStruct.)))
+
+(defn mk-fields-grouping
+ [fields]
+ (Grouping/fields fields))
+
+(defn mk-global-grouping
+ []
+ (mk-fields-grouping []))
+
+(defn mk-direct-grouping
+ []
+ (Grouping/direct (NullStruct.)))
+
+(defn mk-all-grouping
+ []
+ (Grouping/all (NullStruct.)))
+
+(defn mk-none-grouping
+ []
+ (Grouping/none (NullStruct.)))
+
+(defn deserialized-component-object
+ [^ComponentObject obj]
+ (when (not= (.getSetField obj) ComponentObject$_Fields/SERIALIZED_JAVA)
+ (throw (RuntimeException. "Cannot deserialize non-java-serialized object")))
+ (Utils/javaDeserialize (.get_serialized_java obj) Serializable))
+
+(defn serialize-component-object
+ [obj]
+ (ComponentObject/serialized_java (Utils/javaSerialize obj)))
+
+(defn- mk-grouping
+ [grouping-spec]
+ (cond (nil? grouping-spec)
+ (mk-none-grouping)
+
+ (instance? Grouping grouping-spec)
+ grouping-spec
+
+ (instance? CustomStreamGrouping grouping-spec)
+ (Grouping/custom_serialized (Utils/javaSerialize grouping-spec))
+
+ (instance? JavaObject grouping-spec)
+ (Grouping/custom_object grouping-spec)
+
+ (sequential? grouping-spec)
+ (mk-fields-grouping grouping-spec)
+
+ (= grouping-spec :shuffle)
+ (mk-shuffle-grouping)
+
+ (= grouping-spec :local-or-shuffle)
+ (mk-local-or-shuffle-grouping)
+ (= grouping-spec :none)
+ (mk-none-grouping)
+
+ (= grouping-spec :all)
+ (mk-all-grouping)
+
+ (= grouping-spec :global)
+ (mk-global-grouping)
+
+ (= grouping-spec :direct)
+ (mk-direct-grouping)
+
+ true
+ (throw (IllegalArgumentException.
+ (str grouping-spec " is not a valid grouping")))))
+
+(defn- mk-inputs
+ [inputs]
+ (into {} (for [[stream-id grouping-spec] inputs]
+ [(if (sequential? stream-id)
+ (GlobalStreamId. (first stream-id) (second stream-id))
+ (GlobalStreamId. stream-id Utils/DEFAULT_STREAM_ID))
+ (mk-grouping grouping-spec)])))
+
+(defnk mk-bolt-spec*
+ [inputs bolt outputs :p nil :conf nil]
+ (let [common (mk-plain-component-common (mk-inputs inputs) outputs p :conf conf)]
+ (Bolt. (ComponentObject/serialized_java (Utils/javaSerialize bolt))
+ common)))
+
+(defnk mk-spout-spec
+ [spout :parallelism-hint nil :p nil :conf nil]
+ (let [parallelism-hint (if p p parallelism-hint)]
+ {:obj spout :p parallelism-hint :conf conf}))
+
+(defn- shell-component-params
+ [command script-or-output-spec kwargs]
+ (if (string? script-or-output-spec)
+ [(into-array String [command script-or-output-spec])
+ (first kwargs)
+ (rest kwargs)]
+ [(into-array String command)
+ script-or-output-spec
+ kwargs]))
+
+(defnk mk-bolt-spec
+ [inputs bolt :parallelism-hint nil :p nil :conf nil]
+ (let [parallelism-hint (if p p parallelism-hint)]
+ {:obj bolt :inputs inputs :p parallelism-hint :conf conf}))
+
+(defn mk-shell-bolt-spec
+ [inputs command script-or-output-spec & kwargs]
+ (let [[command output-spec kwargs]
+ (shell-component-params command script-or-output-spec kwargs)]
+ (apply mk-bolt-spec inputs
+ (RichShellBolt. command (mk-output-spec output-spec)) kwargs)))
+
+(defn mk-shell-spout-spec
+ [command script-or-output-spec & kwargs]
+ (let [[command output-spec kwargs]
+ (shell-component-params command script-or-output-spec kwargs)]
+ (apply mk-spout-spec
+ (RichShellSpout. command (mk-output-spec output-spec)) kwargs)))
+
+(defn- add-inputs
+ [declarer inputs]
+ (doseq [[id grouping] (mk-inputs inputs)]
+ (.grouping declarer id grouping)))
+
+(defn mk-topology
+ ([spout-map bolt-map]
+ (let [builder (TopologyBuilder.)]
+ (doseq [[name {spout :obj p :p conf :conf}] spout-map]
+ (-> builder (.setSpout name spout (if-not (nil? p) (int p) p)) (.addConfigurations conf)))
+ (doseq [[name {bolt :obj p :p conf :conf inputs :inputs}] bolt-map]
+ (-> builder (.setBolt name bolt (if-not (nil? p) (int p) p)) (.addConfigurations conf) (add-inputs inputs)))
+ (.createTopology builder)))
+ ([spout-map bolt-map state-spout-map]
+ (mk-topology spout-map bolt-map)))
+
+;; clojurify-structure is needed or else every element becomes the same after successive calls
+;; don't know why this happens
+(def STORM-TOPOLOGY-FIELDS
+ (-> StormTopology/metaDataMap clojurify-structure keys))
+
+(def SPOUT-FIELDS
+ [StormTopology$_Fields/SPOUTS
+ StormTopology$_Fields/STATE_SPOUTS])
+
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-clojure/src/test/clj/clojure_test.clj
----------------------------------------------------------------------
diff --git a/storm-clojure/src/test/clj/clojure_test.clj b/storm-clojure/src/test/clj/clojure_test.clj
new file mode 100644
index 0000000..50d3d29
--- /dev/null
+++ b/storm-clojure/src/test/clj/clojure_test.clj
@@ -0,0 +1,158 @@
+;; Licensed to the Apache Software Foundation (ASF) under one
+;; or more contributor license agreements. See the NOTICE file
+;; distributed with this work for additional information
+;; regarding copyright ownership. The ASF licenses this file
+;; to you under the Apache License, Version 2.0 (the
+;; "License"); you may not use this file except in compliance
+;; with the License. You may obtain a copy of the License at
+;;
+;; http://www.apache.org/licenses/LICENSE-2.0
+;;
+;; Unless required by applicable law or agreed to in writing, software
+;; distributed under the License is distributed on an "AS IS" BASIS,
+;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+;; See the License for the specific language governing permissions and
+;; limitations under the License.
+(ns org.apache.storm.clojure-test
+ (:use [clojure test])
+ (:import [org.apache.storm.testing TestWordSpout TestPlannerSpout]
+ [org.apache.storm.tuple Fields])
+ (:use [org.apache.storm testing clojure config])
+ (:use [org.apache.storm.daemon common])
+ (:require [org.apache.storm [thrift :as thrift]])
+ (:import [org.apache.storm Thrift])
+ (:import [org.apache.storm.utils Utils]))
+
+(defbolt lalala-bolt1 ["word"] [[val :as tuple] collector]
+ (let [ret (str val "lalala")]
+ (emit-bolt! collector [ret] :anchor tuple)
+ (ack! collector tuple)
+ ))
+
+(defbolt lalala-bolt2 ["word"] {:prepare true}
+ [conf context collector]
+ (let [state (atom nil)]
+ (reset! state "lalala")
+ (bolt
+ (execute [tuple]
+ (let [ret (-> (.getValue tuple 0) (str @state))]
+ (emit-bolt! collector [ret] :anchor tuple)
+ (ack! collector tuple)
+ ))
+ )))
+
+(defbolt lalala-bolt3 ["word"] {:prepare true :params [prefix]}
+ [conf context collector]
+ (let [state (atom nil)]
+ (bolt
+ (prepare [_ _ _]
+ (reset! state (str prefix "lalala")))
+ (execute [{val "word" :as tuple}]
+ (let [ret (-> (.getValue tuple 0) (str @state))]
+ (emit-bolt! collector [ret] :anchor tuple)
+ (ack! collector tuple)
+ )))
+ ))
+
+(deftest test-clojure-bolt
+ (with-simulated-time-local-cluster [cluster :supervisors 4]
+ (let [nimbus (:nimbus cluster)
+ topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails (TestWordSpout. false))}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareShuffleGrouping)}
+ lalala-bolt1)
+ "3" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareLocalOrShuffleGrouping)}
+ lalala-bolt2)
+ "4" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareShuffleGrouping)}
+ (lalala-bolt3 "_nathan_"))}
+ )
+ results (complete-topology cluster
+ topology
+ :mock-sources {"1" [["david"]
+ ["adam"]
+ ]}
+ )]
+ (is (ms= [["davidlalala"] ["adamlalala"]] (read-tuples results "2")))
+ (is (ms= [["davidlalala"] ["adamlalala"]] (read-tuples results "3")))
+ (is (ms= [["david_nathan_lalala"] ["adam_nathan_lalala"]] (read-tuples results "4")))
+ )))
+
+(defbolt punctuator-bolt ["word" "period" "question" "exclamation"]
+ [tuple collector]
+ (if (= (:word tuple) "bar")
+ (do
+ (emit-bolt! collector {:word "bar" :period "bar" :question "bar"
+ "exclamation" "bar"})
+ (ack! collector tuple))
+ (let [ res (assoc tuple :period (str (:word tuple) "."))
+ res (assoc res :exclamation (str (:word tuple) "!"))
+ res (assoc res :question (str (:word tuple) "?")) ]
+ (emit-bolt! collector res)
+ (ack! collector tuple))))
+
+(deftest test-map-emit
+ (with-simulated-time-local-cluster [cluster :supervisors 4]
+ (let [topology (Thrift/buildTopology
+ {"words" (Thrift/prepareSpoutDetails (TestWordSpout. false))}
+ {"out" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "words" nil)
+ (Thrift/prepareShuffleGrouping)}
+ punctuator-bolt)})
+ results (complete-topology cluster
+ topology
+ :mock-sources {"words" [["foo"] ["bar"]]}
+ )]
+ (is (ms= [["foo" "foo." "foo?" "foo!"]
+ ["bar" "bar" "bar" "bar"]] (read-tuples results "out"))))))
+
+(defbolt conf-query-bolt ["conf" "val"] {:prepare true :params [conf] :conf conf}
+ [conf context collector]
+ (bolt
+ (execute [tuple]
+ (let [name (.getValue tuple 0)
+ val (if (= name "!MAX_MSG_TIMEOUT") (.maxTopologyMessageTimeout context) (get conf name))]
+ (emit-bolt! collector [name val] :anchor tuple)
+ (ack! collector tuple))
+ )))
+
+(deftest test-component-specific-config-clojure
+ (with-simulated-time-local-cluster [cluster]
+ (let [topology (Thrift/buildTopology
+ {"1" (Thrift/prepareSpoutDetails
+ (TestPlannerSpout. (Fields. ["conf"]))
+ nil
+ {TOPOLOGY-MESSAGE-TIMEOUT-SECS 40})}
+ {"2" (Thrift/prepareBoltDetails
+ {(Utils/getGlobalStreamId "1" nil)
+ (Thrift/prepareShuffleGrouping)}
+ (conf-query-bolt {"fake.config" 1
+ TOPOLOGY-MAX-TASK-PARALLELISM 2
+ TOPOLOGY-MAX-SPOUT-PENDING 10})
+ nil
+ {TOPOLOGY-MAX-SPOUT-PENDING 3})})
+ results (complete-topology cluster
+ topology
+ :topology-name "test123"
+ :storm-conf {TOPOLOGY-MAX-TASK-PARALLELISM 10
+ TOPOLOGY-MESSAGE-TIMEOUT-SECS 30}
+ :mock-sources {"1" [["fake.config"]
+ [TOPOLOGY-MAX-TASK-PARALLELISM]
+ [TOPOLOGY-MAX-SPOUT-PENDING]
+ ["!MAX_MSG_TIMEOUT"]
+ [TOPOLOGY-NAME]
+ ]})]
+ (is (= {"fake.config" 1
+ TOPOLOGY-MAX-TASK-PARALLELISM 2
+ TOPOLOGY-MAX-SPOUT-PENDING 3
+ "!MAX_MSG_TIMEOUT" 40
+ TOPOLOGY-NAME "test123"}
+ (->> (read-tuples results "2")
+ (apply concat)
+ (apply hash-map))
+ )))))
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/clj/org/apache/storm/clojure.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/clojure.clj b/storm-core/src/clj/org/apache/storm/clojure.clj
deleted file mode 100644
index 9e1836f..0000000
--- a/storm-core/src/clj/org/apache/storm/clojure.clj
+++ /dev/null
@@ -1,207 +0,0 @@
-;; Licensed to the Apache Software Foundation (ASF) under one
-;; or more contributor license agreements. See the NOTICE file
-;; distributed with this work for additional information
-;; regarding copyright ownership. The ASF licenses this file
-;; to you under the Apache License, Version 2.0 (the
-;; "License"); you may not use this file except in compliance
-;; with the License. You may obtain a copy of the License at
-;;
-;; http://www.apache.org/licenses/LICENSE-2.0
-;;
-;; Unless required by applicable law or agreed to in writing, software
-;; distributed under the License is distributed on an "AS IS" BASIS,
-;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-;; See the License for the specific language governing permissions and
-;; limitations under the License.
-
-(ns org.apache.storm.clojure
- (:use [org.apache.storm util])
- (:import [org.apache.storm StormSubmitter])
- (:import [org.apache.storm.generated StreamInfo])
- (:import [org.apache.storm.tuple Tuple])
- (:import [org.apache.storm.task OutputCollector IBolt TopologyContext])
- (:import [org.apache.storm.spout SpoutOutputCollector ISpout])
- (:import [org.apache.storm.utils Utils])
- (:import [org.apache.storm.clojure ClojureBolt ClojureSpout])
- (:import [java.util Collection List])
- (:require [org.apache.storm [thrift :as thrift]]))
-
-(defn direct-stream [fields]
- (StreamInfo. fields true))
-
-(defn to-spec [avar]
- (let [m (meta avar)]
- [(str (:ns m)) (str (:name m))]))
-
-(defn clojure-bolt* [output-spec fn-var conf-fn-var args]
- (ClojureBolt. (to-spec fn-var) (to-spec conf-fn-var) args (thrift/mk-output-spec output-spec)))
-
-(defmacro clojure-bolt [output-spec fn-sym conf-fn-sym args]
- `(clojure-bolt* ~output-spec (var ~fn-sym) (var ~conf-fn-sym) ~args))
-
-(defn clojure-spout* [output-spec fn-var conf-var args]
- (let [m (meta fn-var)]
- (ClojureSpout. (to-spec fn-var) (to-spec conf-var) args (thrift/mk-output-spec output-spec))
- ))
-
-(defmacro clojure-spout [output-spec fn-sym conf-sym args]
- `(clojure-spout* ~output-spec (var ~fn-sym) (var ~conf-sym) ~args))
-
-(defn normalize-fns [body]
- (for [[name args & impl] body
- :let [args (-> "this"
- gensym
- (cons args)
- vec)]]
- (concat [name args] impl)
- ))
-
-(defmacro bolt [& body]
- (let [[bolt-fns other-fns] (split-with #(not (symbol? %)) body)
- fns (normalize-fns bolt-fns)]
- `(reify IBolt
- ~@fns
- ~@other-fns)))
-
-(defmacro bolt-execute [& body]
- `(bolt
- (~'execute ~@body)))
-
-(defmacro spout [& body]
- (let [[spout-fns other-fns] (split-with #(not (symbol? %)) body)
- fns (normalize-fns spout-fns)]
- `(reify ISpout
- ~@fns
- ~@other-fns)))
-
-(defmacro defbolt [name output-spec & [opts & impl :as all]]
- (if-not (map? opts)
- `(defbolt ~name ~output-spec {} ~@all)
- (let [worker-name (symbol (str name "__"))
- conf-fn-name (symbol (str name "__conf__"))
- params (:params opts)
- conf-code (:conf opts)
- fn-body (if (:prepare opts)
- (cons 'fn impl)
- (let [[args & impl-body] impl
- coll-sym (nth args 1)
- args (vec (take 1 args))
- prepargs [(gensym "conf") (gensym "context") coll-sym]]
- `(fn ~prepargs (bolt (~'execute ~args ~@impl-body)))))
- definer (if params
- `(defn ~name [& args#]
- (clojure-bolt ~output-spec ~worker-name ~conf-fn-name args#))
- `(def ~name
- (clojure-bolt ~output-spec ~worker-name ~conf-fn-name []))
- )
- ]
- `(do
- (defn ~conf-fn-name ~(if params params [])
- ~conf-code
- )
- (defn ~worker-name ~(if params params [])
- ~fn-body
- )
- ~definer
- ))))
-
-(defmacro defspout [name output-spec & [opts & impl :as all]]
- (if-not (map? opts)
- `(defspout ~name ~output-spec {} ~@all)
- (let [worker-name (symbol (str name "__"))
- conf-fn-name (symbol (str name "__conf__"))
- params (:params opts)
- conf-code (:conf opts)
- prepare? (:prepare opts)
- prepare? (if (nil? prepare?) true prepare?)
- fn-body (if prepare?
- (cons 'fn impl)
- (let [[args & impl-body] impl
- coll-sym (first args)
- prepargs [(gensym "conf") (gensym "context") coll-sym]]
- `(fn ~prepargs (spout (~'nextTuple [] ~@impl-body)))))
- definer (if params
- `(defn ~name [& args#]
- (clojure-spout ~output-spec ~worker-name ~conf-fn-name args#))
- `(def ~name
- (clojure-spout ~output-spec ~worker-name ~conf-fn-name []))
- )
- ]
- `(do
- (defn ~conf-fn-name ~(if params params [])
- ~conf-code
- )
- (defn ~worker-name ~(if params params [])
- ~fn-body
- )
- ~definer
- ))))
-
-(defprotocol TupleValues
- (tuple-values [values collector stream]))
-
-(extend-protocol TupleValues
- java.util.Map
- (tuple-values [this collector ^String stream]
- (let [^TopologyContext context (:context collector)
- fields (.. context (getThisOutputFields stream) toList) ]
- (vec (map (into
- (empty this) (for [[k v] this]
- [(if (keyword? k) (name k) k) v]))
- fields))))
- java.util.List
- (tuple-values [this collector stream]
- this))
-
-(defn- collectify
- [obj]
- (if (or (sequential? obj) (instance? Collection obj))
- obj
- [obj]))
-
-(defnk emit-bolt! [collector values
- :stream Utils/DEFAULT_STREAM_ID :anchor []]
- (let [^List anchor (collectify anchor)
- values (tuple-values values collector stream) ]
- (.emit ^OutputCollector (:output-collector collector) stream anchor values)
- ))
-
-(defnk emit-direct-bolt! [collector task values
- :stream Utils/DEFAULT_STREAM_ID :anchor []]
- (let [^List anchor (collectify anchor)
- values (tuple-values values collector stream) ]
- (.emitDirect ^OutputCollector (:output-collector collector) task stream anchor values)
- ))
-
-(defn ack! [collector ^Tuple tuple]
- (.ack ^OutputCollector (:output-collector collector) tuple))
-
-(defn fail! [collector ^Tuple tuple]
- (.fail ^OutputCollector (:output-collector collector) tuple))
-
-(defn report-error! [collector ^Tuple tuple]
- (.reportError ^OutputCollector (:output-collector collector) tuple))
-
-(defnk emit-spout! [collector values
- :stream Utils/DEFAULT_STREAM_ID :id nil]
- (let [values (tuple-values values collector stream)]
- (.emit ^SpoutOutputCollector (:output-collector collector) stream values id)))
-
-(defnk emit-direct-spout! [collector task values
- :stream Utils/DEFAULT_STREAM_ID :id nil]
- (let [values (tuple-values values collector stream)]
- (.emitDirect ^SpoutOutputCollector (:output-collector collector) task stream values id)))
-
-(defalias topology thrift/mk-topology)
-(defalias bolt-spec thrift/mk-bolt-spec)
-(defalias spout-spec thrift/mk-spout-spec)
-(defalias shell-bolt-spec thrift/mk-shell-bolt-spec)
-(defalias shell-spout-spec thrift/mk-shell-spout-spec)
-
-(defn submit-remote-topology [name conf topology]
- (StormSubmitter/submitTopology name conf topology))
-
-(defn local-cluster []
- ;; do this to avoid a cyclic dependency of
- ;; LocalCluster -> testing -> nimbus -> bootstrap -> clojure -> LocalCluster
- (eval '(new org.apache.storm.LocalCluster)))
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/clj/org/apache/storm/command/get_errors.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/command/get_errors.clj b/storm-core/src/clj/org/apache/storm/command/get_errors.clj
index 615a5f3..4f83a86 100644
--- a/storm-core/src/clj/org/apache/storm/command/get_errors.clj
+++ b/storm-core/src/clj/org/apache/storm/command/get_errors.clj
@@ -15,7 +15,8 @@
;; limitations under the License.
(ns org.apache.storm.command.get-errors
(:use [clojure.tools.cli :only [cli]])
- (:use [org.apache.storm thrift log])
+ (:use [org.apache.storm log])
+ (:use [org.apache.storm.internal thrift])
(:use [org.apache.storm util])
(:require [org.apache.storm.daemon
[nimbus :as nimbus]
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/clj/org/apache/storm/command/monitor.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/command/monitor.clj b/storm-core/src/clj/org/apache/storm/command/monitor.clj
index 7fa9b2a..4ec49af 100644
--- a/storm-core/src/clj/org/apache/storm/command/monitor.clj
+++ b/storm-core/src/clj/org/apache/storm/command/monitor.clj
@@ -15,7 +15,7 @@
;; limitations under the License.
(ns org.apache.storm.command.monitor
(:use [clojure.tools.cli :only [cli]])
- (:use [org.apache.storm.thrift :only [with-configured-nimbus-connection]])
+ (:use [org.apache.storm.internal.thrift :only [with-configured-nimbus-connection]])
(:import [org.apache.storm.utils Monitor])
(:gen-class)
)
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/clj/org/apache/storm/command/rebalance.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/command/rebalance.clj b/storm-core/src/clj/org/apache/storm/command/rebalance.clj
index 3868091..8428d14 100644
--- a/storm-core/src/clj/org/apache/storm/command/rebalance.clj
+++ b/storm-core/src/clj/org/apache/storm/command/rebalance.clj
@@ -15,7 +15,8 @@
;; limitations under the License.
(ns org.apache.storm.command.rebalance
(:use [clojure.tools.cli :only [cli]])
- (:use [org.apache.storm thrift config log])
+ (:use [org.apache.storm config log])
+ (:use [org.apache.storm.internal thrift])
(:import [org.apache.storm.generated RebalanceOptions])
(:gen-class))
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/clj/org/apache/storm/command/set_log_level.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/command/set_log_level.clj b/storm-core/src/clj/org/apache/storm/command/set_log_level.clj
index 7e1c3c5..6048246 100644
--- a/storm-core/src/clj/org/apache/storm/command/set_log_level.clj
+++ b/storm-core/src/clj/org/apache/storm/command/set_log_level.clj
@@ -15,7 +15,8 @@
;; limitations under the License.
(ns org.apache.storm.command.set-log-level
(:use [clojure.tools.cli :only [cli]])
- (:use [org.apache.storm thrift log])
+ (:use [org.apache.storm log])
+ (:use [org.apache.storm.internal thrift])
(:import [org.apache.logging.log4j Level])
(:import [org.apache.storm.generated LogConfig LogLevel LogLevelAction])
(:gen-class))
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/clj/org/apache/storm/command/shell_submission.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/command/shell_submission.clj b/storm-core/src/clj/org/apache/storm/command/shell_submission.clj
index 0253338..0d29376 100644
--- a/storm-core/src/clj/org/apache/storm/command/shell_submission.clj
+++ b/storm-core/src/clj/org/apache/storm/command/shell_submission.clj
@@ -17,7 +17,7 @@
(:import [org.apache.storm StormSubmitter]
[org.apache.storm.utils Utils]
[org.apache.storm.zookeeper Zookeeper])
- (:use [org.apache.storm thrift util config log zookeeper])
+ (:use [org.apache.storm util config log zookeeper])
(:require [clojure.string :as str])
(:import [org.apache.storm.utils ConfigUtils])
(:gen-class))
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/clj/org/apache/storm/daemon/common.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/daemon/common.clj b/storm-core/src/clj/org/apache/storm/daemon/common.clj
index eb1ec1e..db7fd40 100644
--- a/storm-core/src/clj/org/apache/storm/daemon/common.clj
+++ b/storm-core/src/clj/org/apache/storm/daemon/common.clj
@@ -16,7 +16,7 @@
(ns org.apache.storm.daemon.common
(:use [org.apache.storm log config util])
(:import [org.apache.storm.generated StormTopology
- InvalidTopologyException GlobalStreamId]
+ InvalidTopologyException GlobalStreamId Grouping Grouping$_Fields]
[org.apache.storm.utils Utils ConfigUtils IPredicate ThriftTopologyUtils]
[org.apache.storm.daemon.metrics.reporters PreparableReporter]
[com.codahale.metrics MetricRegistry])
@@ -28,9 +28,11 @@
(:import [org.apache.storm.security.auth IAuthorizer])
(:import [java.io InterruptedIOException]
[org.json.simple JSONValue])
- (:require [clojure.set :as set])
+ (:import [java.util HashMap])
+ (:import [org.apache.storm Thrift])
+ (:require [clojure.set :as set])
(:require [org.apache.storm.daemon.acker :as acker])
- (:require [org.apache.storm.thrift :as thrift])
+ (:require [metrics.reporters.jmx :as jmx])
(:require [metrics.core :refer [default-registry]]))
(defn start-metrics-reporter [reporter conf]
@@ -91,7 +93,7 @@
(defn topology-bases [storm-cluster-state]
(let [active-topologies (.active-storms storm-cluster-state)]
- (into {}
+ (into {}
(dofor [id active-topologies]
[id (.storm-base storm-cluster-state id nil)]
))
@@ -117,12 +119,12 @@
)))))
(defn- validate-ids! [^StormTopology topology]
- (let [sets (map #(.getFieldValue topology %) thrift/STORM-TOPOLOGY-FIELDS)
+ (let [sets (map #(.getFieldValue topology %) (Thrift/getTopologyFields))
offending (apply set/intersection sets)]
(if-not (empty? offending)
(throw (InvalidTopologyException.
(str "Duplicate component ids: " offending))))
- (doseq [f thrift/STORM-TOPOLOGY-FIELDS
+ (doseq [f (Thrift/getTopologyFields)
:let [obj-map (.getFieldValue topology f)]]
(if-not (ThriftTopologyUtils/isWorkerHook f)
(do
@@ -138,7 +140,7 @@
(defn all-components [^StormTopology topology]
(apply merge {}
- (for [f thrift/STORM-TOPOLOGY-FIELDS]
+ (for [f (Thrift/getTopologyFields)]
(if-not (ThriftTopologyUtils/isWorkerHook f)
(.getFieldValue topology f)))))
@@ -151,13 +153,13 @@
(defn validate-basic! [^StormTopology topology]
(validate-ids! topology)
- (doseq [f thrift/SPOUT-FIELDS
+ (doseq [f (Thrift/getSpoutFields)
obj (->> f (.getFieldValue topology) vals)]
(if-not (empty? (-> obj .get_common .get_inputs))
(throw (InvalidTopologyException. "May not declare inputs for a spout"))))
(doseq [[comp-id comp] (all-components topology)
:let [conf (component-conf comp)
- p (-> comp .get_common thrift/parallelism-hint)]]
+ p (-> comp .get_common (Thrift/getParallelismHint))]]
(when (and (> (conf TOPOLOGY-TASKS) 0)
p
(<= p 0))
@@ -178,7 +180,7 @@
(let [source-streams (-> all-components (get source-component-id) .get_common .get_streams)]
(if-not (contains? source-streams source-stream-id)
(throw (InvalidTopologyException. (str "Component: [" id "] subscribes from non-existent stream: [" source-stream-id "] of component [" source-component-id "]")))
- (if (= :fields (thrift/grouping-type grouping))
+ (if (= Grouping$_Fields/FIELDS (Thrift/groupingType grouping))
(let [grouping-fields (set (.get_fields grouping))
source-stream-fields (-> source-streams (get source-stream-id) .get_output_fields set)
diff-fields (set/difference grouping-fields source-stream-fields)]
@@ -190,12 +192,15 @@
spout-ids (.. topology get_spouts keySet)
spout-inputs (apply merge
(for [id spout-ids]
- {[id ACKER-INIT-STREAM-ID] ["id"]}
+ {(Utils/getGlobalStreamId id ACKER-INIT-STREAM-ID)
+ (Thrift/prepareFieldsGrouping ["id"])}
))
bolt-inputs (apply merge
(for [id bolt-ids]
- {[id ACKER-ACK-STREAM-ID] ["id"]
- [id ACKER-FAIL-STREAM-ID] ["id"]}
+ {(Utils/getGlobalStreamId id ACKER-ACK-STREAM-ID)
+ (Thrift/prepareFieldsGrouping ["id"])
+ (Utils/getGlobalStreamId id ACKER-FAIL-STREAM-ID)
+ (Thrift/prepareFieldsGrouping ["id"])}
))]
(merge spout-inputs bolt-inputs)))
@@ -207,29 +212,31 @@
spout-ids (.. topology get_spouts keySet)
spout-inputs (apply merge
(for [id spout-ids]
- {[id EVENTLOGGER-STREAM-ID] ["component-id"]}
+ {(Utils/getGlobalStreamId id EVENTLOGGER-STREAM-ID)
+ (Thrift/prepareFieldsGrouping ["component-id"])}
))
bolt-inputs (apply merge
(for [id bolt-ids]
- {[id EVENTLOGGER-STREAM-ID] ["component-id"]}
+ {(Utils/getGlobalStreamId id EVENTLOGGER-STREAM-ID)
+ (Thrift/prepareFieldsGrouping ["component-id"])}
))]
(merge spout-inputs bolt-inputs)))
(defn add-acker! [storm-conf ^StormTopology ret]
(let [num-executors (if (nil? (storm-conf TOPOLOGY-ACKER-EXECUTORS)) (storm-conf TOPOLOGY-WORKERS) (storm-conf TOPOLOGY-ACKER-EXECUTORS))
- acker-bolt (thrift/mk-bolt-spec* (acker-inputs ret)
- (new org.apache.storm.daemon.acker)
- {ACKER-ACK-STREAM-ID (thrift/direct-output-fields ["id"])
- ACKER-FAIL-STREAM-ID (thrift/direct-output-fields ["id"])
- }
- :p num-executors
- :conf {TOPOLOGY-TASKS num-executors
- TOPOLOGY-TICK-TUPLE-FREQ-SECS (storm-conf TOPOLOGY-MESSAGE-TIMEOUT-SECS)})]
+ acker-bolt (Thrift/prepareSerializedBoltDetails (acker-inputs ret)
+ (new org.apache.storm.daemon.acker)
+ {ACKER-ACK-STREAM-ID (Thrift/directOutputFields ["id"])
+ ACKER-FAIL-STREAM-ID (Thrift/directOutputFields ["id"])
+ }
+ (Integer. num-executors)
+ {TOPOLOGY-TASKS num-executors
+ TOPOLOGY-TICK-TUPLE-FREQ-SECS (storm-conf TOPOLOGY-MESSAGE-TIMEOUT-SECS)})]
(dofor [[_ bolt] (.get_bolts ret)
:let [common (.get_common bolt)]]
(do
- (.put_to_streams common ACKER-ACK-STREAM-ID (thrift/output-fields ["id" "ack-val"]))
- (.put_to_streams common ACKER-FAIL-STREAM-ID (thrift/output-fields ["id"]))
+ (.put_to_streams common ACKER-ACK-STREAM-ID (Thrift/outputFields ["id" "ack-val"]))
+ (.put_to_streams common ACKER-FAIL-STREAM-ID (Thrift/outputFields ["id"]))
))
(dofor [[_ spout] (.get_spouts ret)
:let [common (.get_common spout)
@@ -239,13 +246,13 @@
(do
;; this set up tick tuples to cause timeouts to be triggered
(.set_json_conf common (JSONValue/toJSONString spout-conf))
- (.put_to_streams common ACKER-INIT-STREAM-ID (thrift/output-fields ["id" "init-val" "spout-task"]))
+ (.put_to_streams common ACKER-INIT-STREAM-ID (Thrift/outputFields ["id" "init-val" "spout-task"]))
(.put_to_inputs common
(GlobalStreamId. ACKER-COMPONENT-ID ACKER-ACK-STREAM-ID)
- (thrift/mk-direct-grouping))
+ (Thrift/prepareDirectGrouping))
(.put_to_inputs common
(GlobalStreamId. ACKER-COMPONENT-ID ACKER-FAIL-STREAM-ID)
- (thrift/mk-direct-grouping))
+ (Thrift/prepareDirectGrouping))
))
(.put_to_bolts ret "__acker" acker-bolt)
))
@@ -254,12 +261,12 @@
(doseq [[_ component] (all-components topology)
:let [common (.get_common component)]]
(.put_to_streams common METRICS-STREAM-ID
- (thrift/output-fields ["task-info" "data-points"]))))
+ (Thrift/outputFields ["task-info" "data-points"]))))
(defn add-system-streams! [^StormTopology topology]
(doseq [[_ component] (all-components topology)
:let [common (.get_common component)]]
- (.put_to_streams common SYSTEM-STREAM-ID (thrift/output-fields ["event"]))))
+ (.put_to_streams common SYSTEM-STREAM-ID (Thrift/outputFields ["event"]))))
(defn map-occurrences [afn coll]
@@ -280,7 +287,7 @@
"Generates a list of component ids for each metrics consumer
e.g. [\"__metrics_org.mycompany.MyMetricsConsumer\", ..] "
[storm-conf]
- (->> (get storm-conf TOPOLOGY-METRICS-CONSUMER-REGISTER)
+ (->> (get storm-conf TOPOLOGY-METRICS-CONSUMER-REGISTER)
(map #(get % "class"))
(number-duplicates)
(map #(str Constants/METRICS_COMPONENT_ID_PREFIX %))))
@@ -288,21 +295,22 @@
(defn metrics-consumer-bolt-specs [storm-conf topology]
(let [component-ids-that-emit-metrics (cons SYSTEM-COMPONENT-ID (keys (all-components topology)))
inputs (->> (for [comp-id component-ids-that-emit-metrics]
- {[comp-id METRICS-STREAM-ID] :shuffle})
+ {(Utils/getGlobalStreamId comp-id METRICS-STREAM-ID)
+ (Thrift/prepareShuffleGrouping)})
(into {}))
-
mk-bolt-spec (fn [class arg p]
- (thrift/mk-bolt-spec*
- inputs
- (org.apache.storm.metric.MetricsConsumerBolt. class arg)
- {} :p p :conf {TOPOLOGY-TASKS p}))]
-
+ (Thrift/prepareSerializedBoltDetails
+ inputs
+ (org.apache.storm.metric.MetricsConsumerBolt. class arg)
+ {}
+ (Integer. p)
+ {TOPOLOGY-TASKS p}))]
+
(map
- (fn [component-id register]
+ (fn [component-id register]
[component-id (mk-bolt-spec (get register "class")
(get register "argument")
(or (get register "parallelism.hint") 1))])
-
(metrics-consumer-register-ids storm-conf)
(get storm-conf TOPOLOGY-METRICS-CONSUMER-REGISTER))))
@@ -313,32 +321,32 @@
(defn add-eventlogger! [storm-conf ^StormTopology ret]
(let [num-executors (if (nil? (storm-conf TOPOLOGY-EVENTLOGGER-EXECUTORS)) (storm-conf TOPOLOGY-WORKERS) (storm-conf TOPOLOGY-EVENTLOGGER-EXECUTORS))
- eventlogger-bolt (thrift/mk-bolt-spec* (eventlogger-inputs ret)
- (EventLoggerBolt.)
- {}
- :p num-executors
- :conf {TOPOLOGY-TASKS num-executors
+ eventlogger-bolt (Thrift/prepareSerializedBoltDetails (eventlogger-inputs ret)
+ (EventLoggerBolt.)
+ {}
+ (Integer. num-executors)
+ {TOPOLOGY-TASKS num-executors
TOPOLOGY-TICK-TUPLE-FREQ-SECS (storm-conf TOPOLOGY-MESSAGE-TIMEOUT-SECS)})]
(doseq [[_ component] (all-components ret)
:let [common (.get_common component)]]
- (.put_to_streams common EVENTLOGGER-STREAM-ID (thrift/output-fields (eventlogger-bolt-fields))))
+ (.put_to_streams common EVENTLOGGER-STREAM-ID (Thrift/outputFields (eventlogger-bolt-fields))))
(.put_to_bolts ret EVENTLOGGER-COMPONENT-ID eventlogger-bolt)
))
-(defn add-metric-components! [storm-conf ^StormTopology topology]
+(defn add-metric-components! [storm-conf ^StormTopology topology]
(doseq [[comp-id bolt-spec] (metrics-consumer-bolt-specs storm-conf topology)]
(.put_to_bolts topology comp-id bolt-spec)))
(defn add-system-components! [conf ^StormTopology topology]
- (let [system-bolt-spec (thrift/mk-bolt-spec*
+ (let [system-bolt-spec (Thrift/prepareSerializedBoltDetails
{}
(SystemBolt.)
- {SYSTEM-TICK-STREAM-ID (thrift/output-fields ["rate_secs"])
- METRICS-TICK-STREAM-ID (thrift/output-fields ["interval"])
- CREDENTIALS-CHANGED-STREAM-ID (thrift/output-fields ["creds"])}
- :p 0
- :conf {TOPOLOGY-TASKS 0})]
+ {SYSTEM-TICK-STREAM-ID (Thrift/outputFields ["rate_secs"])
+ METRICS-TICK-STREAM-ID (Thrift/outputFields ["interval"])
+ CREDENTIALS-CHANGED-STREAM-ID (Thrift/outputFields ["creds"])}
+ (Integer. 0)
+ {TOPOLOGY-TASKS 0})]
(.put_to_bolts topology SYSTEM-COMPONENT-ID system-bolt-spec)))
(defn system-topology! [storm-conf ^StormTopology topology]
@@ -361,7 +369,7 @@
(or (nil? (storm-conf TOPOLOGY-EVENTLOGGER-EXECUTORS)) (> (storm-conf TOPOLOGY-EVENTLOGGER-EXECUTORS) 0)))
(defn num-start-executors [component]
- (thrift/parallelism-hint (.get_common component)))
+ (Thrift/getParallelismHint (.get_common component)))
;TODO: when translating this function, you should replace the map-val with a proper for loop HERE
(defn storm-task-info
@@ -404,11 +412,10 @@
(defn mk-authorization-handler [klassname conf]
(let [aznClass (if klassname (Class/forName klassname))
- aznHandler (if aznClass (.newInstance aznClass))]
+ aznHandler (if aznClass (.newInstance aznClass))]
(if aznHandler (.prepare ^IAuthorizer aznHandler conf))
(log-debug "authorization class name:" klassname
" class:" aznClass
" handler:" aznHandler)
aznHandler
- ))
-
+ ))
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/clj/org/apache/storm/daemon/executor.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/daemon/executor.clj b/storm-core/src/clj/org/apache/storm/daemon/executor.clj
index 3af365b..14a2f6e 100644
--- a/storm-core/src/clj/org/apache/storm/daemon/executor.clj
+++ b/storm-core/src/clj/org/apache/storm/daemon/executor.clj
@@ -15,11 +15,11 @@
;; limitations under the License.
(ns org.apache.storm.daemon.executor
(:use [org.apache.storm.daemon common])
- (:import [org.apache.storm.generated Grouping]
+ (:import [org.apache.storm.generated Grouping Grouping$_Fields]
[java.io Serializable])
(:use [org.apache.storm util config log timer stats])
(:import [java.util List Random HashMap ArrayList LinkedList Map])
- (:import [org.apache.storm ICredentialsListener])
+ (:import [org.apache.storm ICredentialsListener Thrift])
(:import [org.apache.storm.hooks ITaskHook])
(:import [org.apache.storm.tuple AddressedTuple Tuple Fields TupleImpl MessageId])
(:import [org.apache.storm.spout ISpoutWaitStrategy ISpout SpoutOutputCollector ISpoutOutputCollector])
@@ -40,8 +40,7 @@
[java.util.concurrent ConcurrentLinkedQueue]
[org.json.simple JSONValue]
[com.lmax.disruptor.dsl ProducerType])
- (:require [org.apache.storm [thrift :as thrift]
- [cluster :as cluster] [stats :as stats]])
+ (:require [org.apache.storm [cluster :as cluster] [stats :as stats]])
(:require [org.apache.storm.daemon [task :as task]])
(:require [org.apache.storm.daemon.builtin-metrics :as builtin-metrics])
(:require [clojure.set :as set]))
@@ -77,38 +76,38 @@
(let [num-tasks (count target-tasks)
random (Random.)
target-tasks (vec (sort target-tasks))]
- (condp = (thrift/grouping-type thrift-grouping)
- :fields
- (if (thrift/global-grouping? thrift-grouping)
+ (condp = (Thrift/groupingType thrift-grouping)
+ Grouping$_Fields/FIELDS
+ (if (Thrift/isGlobalGrouping thrift-grouping)
(fn [task-id tuple load]
;; It's possible for target to have multiple tasks if it reads multiple sources
(first target-tasks))
- (let [group-fields (Fields. (thrift/field-grouping thrift-grouping))]
+ (let [group-fields (Fields. (Thrift/fieldGrouping thrift-grouping))]
(mk-fields-grouper out-fields group-fields target-tasks)
))
- :all
+ Grouping$_Fields/ALL
(fn [task-id tuple load] target-tasks)
- :shuffle
+ Grouping$_Fields/SHUFFLE
(mk-shuffle-grouper target-tasks topo-conf context component-id stream-id)
- :local-or-shuffle
+ Grouping$_Fields/LOCAL_OR_SHUFFLE
(let [same-tasks (set/intersection
(set target-tasks)
(set (.getThisWorkerTasks context)))]
(if-not (empty? same-tasks)
(mk-shuffle-grouper (vec same-tasks) topo-conf context component-id stream-id)
(mk-shuffle-grouper target-tasks topo-conf context component-id stream-id)))
- :none
+ Grouping$_Fields/NONE
(fn [task-id tuple load]
(let [i (mod (.nextInt random) num-tasks)]
(get target-tasks i)
))
- :custom-object
- (let [grouping (thrift/instantiate-java-object (.get_custom_object thrift-grouping))]
+ Grouping$_Fields/CUSTOM_OBJECT
+ (let [grouping (Thrift/instantiateJavaObject (.get_custom_object thrift-grouping))]
(mk-custom-grouper grouping context component-id stream-id target-tasks))
- :custom-serialized
+ Grouping$_Fields/CUSTOM_SERIALIZED
(let [grouping (Utils/javaDeserialize (.get_custom_serialized thrift-grouping) Serializable)]
(mk-custom-grouper grouping context component-id stream-id target-tasks))
- :direct
+ Grouping$_Fields/DIRECT
:direct
)))
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/clj/org/apache/storm/daemon/task.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/daemon/task.clj b/storm-core/src/clj/org/apache/storm/daemon/task.clj
index a097e36..77abdec 100644
--- a/storm-core/src/clj/org/apache/storm/daemon/task.clj
+++ b/storm-core/src/clj/org/apache/storm/daemon/task.clj
@@ -27,8 +27,8 @@
(:import [org.apache.storm.generated ShellComponent JavaObject])
(:import [org.apache.storm.spout ShellSpout])
(:import [java.util Collection List ArrayList])
+ (:import [org.apache.storm Thrift])
(:require [org.apache.storm
- [thrift :as thrift]
[stats :as stats]])
(:require [org.apache.storm.daemon.builtin-metrics :as builtin-metrics]))
@@ -83,7 +83,7 @@
(ShellBolt. obj))
obj )
obj (if (instance? JavaObject obj)
- (thrift/instantiate-java-object obj)
+ (Thrift/instantiateJavaObject obj)
obj )]
obj
))
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/clj/org/apache/storm/internal/clojure.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/internal/clojure.clj b/storm-core/src/clj/org/apache/storm/internal/clojure.clj
new file mode 100644
index 0000000..3f29757
--- /dev/null
+++ b/storm-core/src/clj/org/apache/storm/internal/clojure.clj
@@ -0,0 +1,201 @@
+;; Licensed to the Apache Software Foundation (ASF) under one
+;; or more contributor license agreements. See the NOTICE file
+;; distributed with this work for additional information
+;; regarding copyright ownership. The ASF licenses this file
+;; to you under the Apache License, Version 2.0 (the
+;; "License"); you may not use this file except in compliance
+;; with the License. You may obtain a copy of the License at
+;;
+;; http://www.apache.org/licenses/LICENSE-2.0
+;;
+;; Unless required by applicable law or agreed to in writing, software
+;; distributed under the License is distributed on an "AS IS" BASIS,
+;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+;; See the License for the specific language governing permissions and
+;; limitations under the License.
+
+(ns org.apache.storm.internal.clojure
+ (:use [org.apache.storm util])
+ (:import [org.apache.storm StormSubmitter])
+ (:import [org.apache.storm.generated StreamInfo])
+ (:import [org.apache.storm.tuple Tuple])
+ (:import [org.apache.storm.task OutputCollector IBolt TopologyContext])
+ (:import [org.apache.storm.spout SpoutOutputCollector ISpout])
+ (:import [org.apache.storm.utils Utils])
+ (:import [org.apache.storm.clojure ClojureBolt ClojureSpout])
+ (:import [java.util Collection List])
+ (:require [org.apache.storm.internal [thrift :as thrift]]))
+
+(defn direct-stream [fields]
+ (StreamInfo. fields true))
+
+(defn to-spec [avar]
+ (let [m (meta avar)]
+ [(str (:ns m)) (str (:name m))]))
+
+(defn clojure-bolt* [output-spec fn-var conf-fn-var args]
+ (ClojureBolt. (to-spec fn-var) (to-spec conf-fn-var) args (thrift/mk-output-spec output-spec)))
+
+(defmacro clojure-bolt [output-spec fn-sym conf-fn-sym args]
+ `(clojure-bolt* ~output-spec (var ~fn-sym) (var ~conf-fn-sym) ~args))
+
+(defn clojure-spout* [output-spec fn-var conf-var args]
+ (let [m (meta fn-var)]
+ (ClojureSpout. (to-spec fn-var) (to-spec conf-var) args (thrift/mk-output-spec output-spec))
+ ))
+
+(defmacro clojure-spout [output-spec fn-sym conf-sym args]
+ `(clojure-spout* ~output-spec (var ~fn-sym) (var ~conf-sym) ~args))
+
+(defn normalize-fns [body]
+ (for [[name args & impl] body
+ :let [args (-> "this"
+ gensym
+ (cons args)
+ vec)]]
+ (concat [name args] impl)
+ ))
+
+(defmacro bolt [& body]
+ (let [[bolt-fns other-fns] (split-with #(not (symbol? %)) body)
+ fns (normalize-fns bolt-fns)]
+ `(reify IBolt
+ ~@fns
+ ~@other-fns)))
+
+(defmacro bolt-execute [& body]
+ `(bolt
+ (~'execute ~@body)))
+
+(defmacro spout [& body]
+ (let [[spout-fns other-fns] (split-with #(not (symbol? %)) body)
+ fns (normalize-fns spout-fns)]
+ `(reify ISpout
+ ~@fns
+ ~@other-fns)))
+
+(defmacro defbolt [name output-spec & [opts & impl :as all]]
+ (if-not (map? opts)
+ `(defbolt ~name ~output-spec {} ~@all)
+ (let [worker-name (symbol (str name "__"))
+ conf-fn-name (symbol (str name "__conf__"))
+ params (:params opts)
+ conf-code (:conf opts)
+ fn-body (if (:prepare opts)
+ (cons 'fn impl)
+ (let [[args & impl-body] impl
+ coll-sym (nth args 1)
+ args (vec (take 1 args))
+ prepargs [(gensym "conf") (gensym "context") coll-sym]]
+ `(fn ~prepargs (bolt (~'execute ~args ~@impl-body)))))
+ definer (if params
+ `(defn ~name [& args#]
+ (clojure-bolt ~output-spec ~worker-name ~conf-fn-name args#))
+ `(def ~name
+ (clojure-bolt ~output-spec ~worker-name ~conf-fn-name []))
+ )
+ ]
+ `(do
+ (defn ~conf-fn-name ~(if params params [])
+ ~conf-code
+ )
+ (defn ~worker-name ~(if params params [])
+ ~fn-body
+ )
+ ~definer
+ ))))
+
+(defmacro defspout [name output-spec & [opts & impl :as all]]
+ (if-not (map? opts)
+ `(defspout ~name ~output-spec {} ~@all)
+ (let [worker-name (symbol (str name "__"))
+ conf-fn-name (symbol (str name "__conf__"))
+ params (:params opts)
+ conf-code (:conf opts)
+ prepare? (:prepare opts)
+ prepare? (if (nil? prepare?) true prepare?)
+ fn-body (if prepare?
+ (cons 'fn impl)
+ (let [[args & impl-body] impl
+ coll-sym (first args)
+ prepargs [(gensym "conf") (gensym "context") coll-sym]]
+ `(fn ~prepargs (spout (~'nextTuple [] ~@impl-body)))))
+ definer (if params
+ `(defn ~name [& args#]
+ (clojure-spout ~output-spec ~worker-name ~conf-fn-name args#))
+ `(def ~name
+ (clojure-spout ~output-spec ~worker-name ~conf-fn-name []))
+ )
+ ]
+ `(do
+ (defn ~conf-fn-name ~(if params params [])
+ ~conf-code
+ )
+ (defn ~worker-name ~(if params params [])
+ ~fn-body
+ )
+ ~definer
+ ))))
+
+(defprotocol TupleValues
+ (tuple-values [values collector stream]))
+
+(extend-protocol TupleValues
+ java.util.Map
+ (tuple-values [this collector ^String stream]
+ (let [^TopologyContext context (:context collector)
+ fields (.. context (getThisOutputFields stream) toList) ]
+ (vec (map (into
+ (empty this) (for [[k v] this]
+ [(if (keyword? k) (name k) k) v]))
+ fields))))
+ java.util.List
+ (tuple-values [this collector stream]
+ this))
+
+(defn- collectify
+ [obj]
+ (if (or (sequential? obj) (instance? Collection obj))
+ obj
+ [obj]))
+
+(defnk emit-bolt! [collector values
+ :stream Utils/DEFAULT_STREAM_ID :anchor []]
+ (let [^List anchor (collectify anchor)
+ values (tuple-values values collector stream) ]
+ (.emit ^OutputCollector (:output-collector collector) stream anchor values)
+ ))
+
+(defnk emit-direct-bolt! [collector task values
+ :stream Utils/DEFAULT_STREAM_ID :anchor []]
+ (let [^List anchor (collectify anchor)
+ values (tuple-values values collector stream) ]
+ (.emitDirect ^OutputCollector (:output-collector collector) task stream anchor values)
+ ))
+
+(defn ack! [collector ^Tuple tuple]
+ (.ack ^OutputCollector (:output-collector collector) tuple))
+
+(defn fail! [collector ^Tuple tuple]
+ (.fail ^OutputCollector (:output-collector collector) tuple))
+
+(defn report-error! [collector ^Tuple tuple]
+ (.reportError ^OutputCollector (:output-collector collector) tuple))
+
+(defnk emit-spout! [collector values
+ :stream Utils/DEFAULT_STREAM_ID :id nil]
+ (let [values (tuple-values values collector stream)]
+ (.emit ^SpoutOutputCollector (:output-collector collector) stream values id)))
+
+(defnk emit-direct-spout! [collector task values
+ :stream Utils/DEFAULT_STREAM_ID :id nil]
+ (let [values (tuple-values values collector stream)]
+ (.emitDirect ^SpoutOutputCollector (:output-collector collector) task stream values id)))
+
+(defn submit-remote-topology [name conf topology]
+ (StormSubmitter/submitTopology name conf topology))
+
+(defn local-cluster []
+ ;; do this to avoid a cyclic dependency of
+ ;; LocalCluster -> testing -> nimbus -> bootstrap -> clojure -> LocalCluster
+ (eval '(new org.apache.storm.LocalCluster)))
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/clj/org/apache/storm/internal/thrift.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/internal/thrift.clj b/storm-core/src/clj/org/apache/storm/internal/thrift.clj
new file mode 100644
index 0000000..4ccf8a7
--- /dev/null
+++ b/storm-core/src/clj/org/apache/storm/internal/thrift.clj
@@ -0,0 +1,96 @@
+;; Licensed to the Apache Software Foundation (ASF) under one
+;; or more contributor license agreements. See the NOTICE file
+;; distributed with this work for additional information
+;; regarding copyright ownership. The ASF licenses this file
+;; to you under the Apache License, Version 2.0 (the
+;; "License"); you may not use this file except in compliance
+;; with the License. You may obtain a copy of the License at
+;;
+;; http://www.apache.org/licenses/LICENSE-2.0
+;;
+;; Unless required by applicable law or agreed to in writing, software
+;; distributed under the License is distributed on an "AS IS" BASIS,
+;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+;; See the License for the specific language governing permissions and
+;; limitations under the License.
+
+(ns org.apache.storm.internal.thrift
+ (:import [java.util HashMap]
+ [java.io Serializable]
+ [org.apache.storm.generated NodeInfo Assignment])
+ (:import [org.apache.storm.generated JavaObject Grouping Nimbus StormTopology
+ StormTopology$_Fields Bolt Nimbus$Client Nimbus$Iface
+ ComponentCommon Grouping$_Fields SpoutSpec NullStruct StreamInfo
+ GlobalStreamId ComponentObject ComponentObject$_Fields
+ ShellComponent SupervisorInfo])
+ (:import [org.apache.storm.utils Utils NimbusClient ConfigUtils])
+ (:import [org.apache.storm Constants])
+ (:import [org.apache.storm.security.auth ReqContext])
+ (:import [org.apache.storm.grouping CustomStreamGrouping])
+ (:import [org.apache.storm.topology TopologyBuilder])
+ (:import [org.apache.storm.clojure RichShellBolt RichShellSpout])
+ (:import [org.apache.thrift.transport TTransport])
+ (:use [org.apache.storm util config log zookeeper]))
+
+;; Leaving this definition as core.clj is using them as a nested keyword argument
+;; Must remove once core.clj is ported to java
+(def grouping-constants
+ {Grouping$_Fields/FIELDS :fields
+ Grouping$_Fields/SHUFFLE :shuffle
+ Grouping$_Fields/ALL :all
+ Grouping$_Fields/NONE :none
+ Grouping$_Fields/CUSTOM_SERIALIZED :custom-serialized
+ Grouping$_Fields/CUSTOM_OBJECT :custom-object
+ Grouping$_Fields/DIRECT :direct
+ Grouping$_Fields/LOCAL_OR_SHUFFLE :local-or-shuffle})
+
+;; Leaving this method as core.clj is using them as a nested keyword argument
+;; Must remove once core.clj is ported to java
+(defn grouping-type
+ [^Grouping grouping]
+ (grouping-constants (.getSetField grouping)))
+
+(defn nimbus-client-and-conn
+ ([host port]
+ (nimbus-client-and-conn host port nil))
+ ([host port as-user]
+ (log-message "Connecting to Nimbus at " host ":" port " as user: " as-user)
+ (let [conf (clojurify-structure (ConfigUtils/readStormConfig))
+ nimbusClient (NimbusClient. conf host port nil as-user)
+ client (.getClient nimbusClient)
+ transport (.transport nimbusClient)]
+ [client transport] )))
+
+(defmacro with-nimbus-connection
+ [[client-sym host port] & body]
+ `(let [[^Nimbus$Client ~client-sym ^TTransport conn#] (nimbus-client-and-conn ~host ~port)]
+ (try
+ ~@body
+ (finally (.close conn#)))))
+
+(defmacro with-configured-nimbus-connection
+ [client-sym & body]
+ `(let [conf# (clojurify-structure (ConfigUtils/readStormConfig))
+ context# (ReqContext/context)
+ user# (if (.principal context#) (.getName (.principal context#)))
+ nimbusClient# (NimbusClient/getConfiguredClientAs conf# user#)
+ ~client-sym (.getClient nimbusClient#)
+ conn# (.transport nimbusClient#)
+ ]
+ (try
+ ~@body
+ (finally (.close conn#)))))
+
+;; Leaving this definition as core.clj is using them as a nested keyword argument
+;; Must remove once core.clj is ported to java
+(defn mk-output-spec
+ [output-spec]
+ (let [output-spec (if (map? output-spec)
+ output-spec
+ {Utils/DEFAULT_STREAM_ID output-spec})]
+ (map-val
+ (fn [out]
+ (if (instance? StreamInfo out)
+ out
+ (StreamInfo. out false)))
+ output-spec)))
http://git-wip-us.apache.org/repos/asf/storm/blob/de9cb106/storm-core/src/clj/org/apache/storm/testing.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/testing.clj b/storm-core/src/clj/org/apache/storm/testing.clj
index c872742..4ad5ff8 100644
--- a/storm-core/src/clj/org/apache/storm/testing.clj
+++ b/storm-core/src/clj/org/apache/storm/testing.clj
@@ -44,13 +44,15 @@
(:import [org.apache.storm.transactional TransactionalSpoutCoordinator])
(:import [org.apache.storm.transactional.partitioned PartitionedTransactionalSpoutExecutor])
(:import [org.apache.storm.tuple Tuple])
+ (:import [org.apache.storm Thrift])
(:import [org.apache.storm.generated StormTopology])
(:import [org.apache.storm.task TopologyContext]
(org.apache.storm.messaging IContext)
[org.json.simple JSONValue])
(:require [org.apache.storm [zookeeper :as zk]])
(:require [org.apache.storm.daemon.acker :as acker])
- (:use [org.apache.storm cluster util thrift config log local-state]))
+ (:use [org.apache.storm cluster util config log local-state])
+ (:use [org.apache.storm.internal thrift]))
(defn feeder-spout
[fields]
@@ -526,7 +528,7 @@
(for [[_ spout-spec] spec-map]
(-> spout-spec
.get_spout_object
- deserialized-component-object)))
+ (Thrift/deserializeComponentObject))))
(defn capture-topology
[topology]
@@ -543,11 +545,11 @@
(assoc (clojurify-structure bolts)
(Utils/uuid)
(Bolt.
- (serialize-component-object capturer)
- (mk-plain-component-common (into {} (for [[id direct?] all-streams]
+ (Thrift/serializeComponentObject capturer)
+ (Thrift/prepareComponentCommon (into {} (for [[id direct?] all-streams]
[id (if direct?
- (mk-direct-grouping)
- (mk-global-grouping))]))
+ (Thrift/prepareDirectGrouping)
+ (Thrift/prepareGlobalGrouping))]))
{}
nil))))
{:topology topology
@@ -577,7 +579,7 @@
mock-sources)]
(doseq [[id spout] replacements]
(let [spout-spec (get spouts id)]
- (.set_spout_object spout-spec (serialize-component-object spout))))
+ (.set_spout_object spout-spec (Thrift/serializeComponentObject spout))))
(doseq [spout (spout-objects spouts)]
(when-not (extends? CompletableSpout (.getClass spout))
(throw (RuntimeException. (str "Cannot complete topology unless every spout is a CompletableSpout (or mocked to be); failed by " spout)))))
@@ -636,12 +638,12 @@
(let [track-id (::track-id tracked-cluster)
ret (.deepCopy topology)]
(dofor [[_ bolt] (.get_bolts ret)
- :let [obj (deserialized-component-object (.get_bolt_object bolt))]]
- (.set_bolt_object bolt (serialize-component-object
+ :let [obj (Thrift/deserializeComponentObject (.get_bolt_object bolt))]]
+ (.set_bolt_object bolt (Thrift/serializeComponentObject
(BoltTracker. obj track-id))))
(dofor [[_ spout] (.get_spouts ret)
- :let [obj (deserialized-component-object (.get_spout_object spout))]]
- (.set_spout_object spout (serialize-component-object
+ :let [obj (Thrift/deserializeComponentObject (.get_spout_object spout))]]
+ (.set_spout_object spout (Thrift/serializeComponentObject
(SpoutTracker. obj track-id))))
{:topology ret
:last-spout-emit (atom 0)
@@ -723,8 +725,9 @@
(->> (iterate inc 1)
(take (count values))
(map #(str "field" %))))
- spout-spec (mk-spout-spec* (TestWordSpout.)
- {stream fields})
+ spout-spec (Thrift/prepareSerializedSpoutDetails
+ (TestWordSpout.)
+ {stream fields})
topology (StormTopology. {component spout-spec} {} {})
context (TopologyContext.
topology
[4/5] storm git commit: Merge branch 'thrift-backport' of
https://github.com/redsanket/storm into STORM-1258
Posted by bo...@apache.org.
Merge branch 'thrift-backport' of https://github.com/redsanket/storm into STORM-1258
STORM-1258: port thrift.clj to Thrift.java
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/5e4c0a74
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/5e4c0a74
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/5e4c0a74
Branch: refs/heads/master
Commit: 5e4c0a74dbda21d45b2206e929b173fa14833d71
Parents: bb60f11 de9cb10
Author: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Authored: Thu Feb 18 10:00:19 2016 -0600
Committer: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Committed: Thu Feb 18 10:00:19 2016 -0600
----------------------------------------------------------------------
examples/storm-starter/pom.xml | 10 +
.../org/apache/storm/starter/clj/word_count.clj | 3 +-
pom.xml | 1 +
storm-clojure/pom.xml | 74 ++++
.../src/clj/org/apache/storm/clojure.clj | 207 +++++++++++
.../src/clj/org/apache/storm/thrift.clj | 286 +++++++++++++++
storm-clojure/src/test/clj/clojure_test.clj | 158 +++++++++
storm-core/src/clj/org/apache/storm/clojure.clj | 207 -----------
.../clj/org/apache/storm/command/get_errors.clj | 3 +-
.../clj/org/apache/storm/command/monitor.clj | 2 +-
.../clj/org/apache/storm/command/rebalance.clj | 3 +-
.../org/apache/storm/command/set_log_level.clj | 3 +-
.../apache/storm/command/shell_submission.clj | 2 +-
.../src/clj/org/apache/storm/daemon/common.clj | 121 ++++---
.../clj/org/apache/storm/daemon/executor.clj | 31 +-
.../src/clj/org/apache/storm/daemon/task.clj | 4 +-
.../clj/org/apache/storm/internal/clojure.clj | 201 +++++++++++
.../clj/org/apache/storm/internal/thrift.clj | 96 +++++
storm-core/src/clj/org/apache/storm/testing.clj | 29 +-
storm-core/src/clj/org/apache/storm/thrift.clj | 286 ---------------
storm-core/src/clj/org/apache/storm/ui/core.clj | 2 +-
storm-core/src/jvm/org/apache/storm/Thrift.java | 351 +++++++++++++++++++
.../jvm/org/apache/storm/testing/NGrouping.java | 4 +-
.../storm/testing/PythonShellMetricsBolt.java | 14 +-
.../storm/testing/PythonShellMetricsSpout.java | 8 +-
.../src/jvm/org/apache/storm/utils/Utils.java | 7 +
.../org/apache/storm/integration_test.clj | 259 +++++++-------
.../org/apache/storm/testing4j_test.clj | 72 ++--
.../test/clj/org/apache/storm/clojure_test.clj | 64 ++--
.../test/clj/org/apache/storm/cluster_test.clj | 3 +-
.../test/clj/org/apache/storm/drpc_test.clj | 23 +-
.../test/clj/org/apache/storm/grouping_test.clj | 56 +--
.../storm/messaging/netty_integration_test.clj | 18 +-
.../clj/org/apache/storm/messaging_test.clj | 14 +-
.../test/clj/org/apache/storm/metrics_test.clj | 85 +++--
.../test/clj/org/apache/storm/nimbus_test.clj | 257 +++++++++-----
.../scheduler/resource_aware_scheduler_test.clj | 3 +-
.../clj/org/apache/storm/supervisor_test.clj | 154 ++++----
.../clj/org/apache/storm/tick_tuple_test.clj | 15 +-
.../clj/org/apache/storm/transactional_test.clj | 3 +-
40 files changed, 2128 insertions(+), 1011 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/5e4c0a74/pom.xml
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/5e4c0a74/storm-core/src/jvm/org/apache/storm/utils/Utils.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/5e4c0a74/storm-core/test/clj/org/apache/storm/supervisor_test.clj
----------------------------------------------------------------------