Permalink
Browse files

finish INimbus/ISupervisor integration

  • Loading branch information...
1 parent d5376e5 commit d74ae4367e7e7204df298d7918b1f2a277b9da67 @nathanmarz committed Apr 5, 2012
View
4 src/clj/backtype/storm/daemon/nimbus.clj
@@ -427,10 +427,10 @@
(defn compute-new-task->node+port [nimbus ^TopologyDetails topology-details existing-assignment callback scratch?]
(let [conf (:conf nimbus)
storm-cluster-state (:storm-cluster-state nimbus)
- task-heartbeat-cache (:task-heartbeats-cache nimbus)
+ task-heartbeats-cache (:task-heartbeats-cache nimbus)
storm-id (.getId topology-details)
- available-slots (available-slots nimbus callback)
+ available-slots (available-slots nimbus callback topology-details)
storm-conf (read-storm-conf conf storm-id)
all-task-ids (set (.task-ids storm-cluster-state storm-id))
View
2 src/clj/backtype/storm/daemon/supervisor.clj
@@ -281,7 +281,7 @@
))
(log-debug "Writing new assignment "
(pr-str new-assignment))
- (doseq [p (set/difference (set (keys existing-assigned))
+ (doseq [p (set/difference (set (keys existing-assignment))
(set (keys new-assignment)))]
(.killedWorker isupervisor (int p)))
(.put local-state
View
12 src/jvm/backtype/storm/scheduler/TopologyDetails.java
@@ -1,18 +1,18 @@
package backtype.storm.scheduler;
-import backtype.storm.task.GeneralTopologyContext;
+import backtype.storm.generated.StormTopology;
import java.util.Map;
public class TopologyDetails {
String topologyId;
Map topologyConf;
- GeneralTopologyContext context;
+ StormTopology topology;
- public TopologyDetails(String topologyId, Map topologyConf, GeneralTopologyContext context) {
+ public TopologyDetails(String topologyId, Map topologyConf, StormTopology topology) {
this.topologyId = topologyId;
this.topologyConf = topologyConf;
- this.context = context;
+ this.topology = topology;
}
public String getId() {
@@ -23,7 +23,7 @@ public Map getConf() {
return topologyConf;
}
- public GeneralTopologyContext getContext() {
- return context;
+ public StormTopology getTopology() {
+ return topology;
}
}
View
2 test/clj/backtype/storm/nimbus_test.clj
@@ -417,7 +417,7 @@
(bind storm-id2 (get-storm-id cluster-state "t2"))
(.shutdown nimbus)
(rmr (master-stormdist-root conf storm-id1))
- (bind nimbus (nimbus/service-handler conf) (nimbus/standalone-nimbus))
+ (bind nimbus (nimbus/service-handler conf (nimbus/standalone-nimbus)))
(is ( = #{storm-id2} (set (.active-storms cluster-state))))
(.shutdown nimbus)
(.disconnect cluster-state)
View
2 test/clj/backtype/storm/supervisor_test.clj
@@ -197,7 +197,7 @@
(bind sup1 (add-supervisor cluster :id "sup" :ports [5 6 7]))
(advance-cluster-time cluster 4)
(bind hb (get-heartbeat cluster "sup"))
- (is (= #{5 6 7} (set (:worker-ports hb))))
+ (is (= #{5 6 7} (set (:meta hb))))
(check-heartbeat cluster "sup" 3)
(advance-cluster-time cluster 3)
(check-heartbeat cluster "sup" 3)

0 comments on commit d74ae43

Please sign in to comment.