xumingming

(try ....
  (catch Exception e
    (println e)))

xumingming

;;; See the inspirational SO question: http://stackoverflow.com/questions/3346382

(require 'clojure.contrib.trace)

(defn trace-ns
  "Replaces each function from the given namespace with a version wrapped
  in a tracing call. Can be undone with untrace-ns. ns should be a namespace
  object or a symbol."
  [ns]
  (doseq [s (keys (ns-interns ns))

xumingming

(defmacro dbg[x] `(let [x# ~x] (println "dbg:" '~x "=" x#) x#))

xumingming

Testing backtype.storm.integration-test
0    [main] INFO  backtype.storm.zookeeper  - Starting inprocess zookeeper at port 2000 and dir /tmp/5980f969-535a-4cc3-808d-6f47bcdb45e7
782  [main] INFO  backtype.storm.daemon.nimbus  - Starting Nimbus with conf {"topology.fall.back.on.java.serialization" true, "zmq.linger.millis" 0, "topology.skip.missing.kryo.registrations" true, "ui.childopts" "-Xmx768m", "storm.zookeeper.session.timeout" 20000, "nimbus.reassign" true, "nimbus.monitor.freq.secs" 10, "java.library.path" "/usr/local/lib:/opt/local/lib:/usr/lib", "storm.local.dir" "/tmp/feceb475-b8e8-4b84-beeb-ccc44581d0e0", "supervisor.worker.start.timeout.secs" 240, "nimbus.cleanup.inbox.freq.secs" 600, "nimbus.inbox.jar.expiration.secs" 3600, "storm.zookeeper.port" 2000, "transactional.zookeeper.port" nil, "transactional.zookeeper.servers" nil, "storm.zookeeper.root" "/storm", "supervisor.enable" true, "storm.zookeeper.servers" ["localhost"], "transactional.zookeeper.root" "/transactional", "topology.worker.childo

xumingming

Testing backtype.storm.integration-test
0    [main] INFO  backtype.storm.zookeeper  - Starting inprocess zookeeper at port 2000 and dir /tmp/d58bdfed-90e7-40e5-9e1e-8cc9c568a734
1301 [main] INFO  backtype.storm.daemon.nimbus  - Starting Nimbus with conf {"topology.fall.back.on.java.serialization" true, "zmq.linger.millis" 0, "topology.skip.missing.kryo.registrations" true, "ui.childopts" "-Xmx768m", "storm.zookeeper.session.timeout" 20000, "nimbus.reassign" true, "nimbus.monitor.freq.secs" 10, "java.library.path" "/usr/local/lib:/opt/local/lib:/usr/lib", "storm.local.dir" "/tmp/fb314afc-25da-43ad-9dab-af60ca526096", "supervisor.worker.start.timeout.secs" 240, "nimbus.cleanup.inbox.freq.secs" 600, "nimbus.inbox.jar.expiration.secs" 3600, "storm.zookeeper.port" 2000, "transactional.zookeeper.port" nil, "transactional.zookeeper.servers" nil, "storm.zookeeper.root" "/storm", "supervisor.enable" true, "storm.zookeeper.servers" ["localhost"], "transactional.zookeeper.root" "/transactional", "topology.worker.childop

xumingming

Testing backtype.storm.integration-test
1    [main] INFO  backtype.storm.zookeeper  - Starting inprocess zookeeper at port 2000 and dir /tmp/f31cd371-6665-4694-ba53-ef8069530521
1218 [main] INFO  backtype.storm.daemon.nimbus  - Starting Nimbus with conf {"topology.fall.back.on.java.serialization" true, "zmq.linger.millis" 0, "topology.skip.missing.kryo.registrations" true, "ui.childopts" "-Xmx768m", "storm.zookeeper.session.timeout" 20000, "nimbus.reassign" true, "nimbus.monitor.freq.secs" 10, "java.library.path" "/usr/local/lib:/opt/local/lib:/usr/lib", "storm.local.dir" "/tmp/083bd09f-7709-462d-9525-aba269e463e8", "supervisor.worker.start.timeout.secs" 240, "nimbus.cleanup.inbox.freq.secs" 600, "nimbus.inbox.jar.expiration.secs" 3600, "storm.zookeeper.port" 2000, "transactional.zookeeper.port" nil, "transactional.zookeeper.servers" nil, "storm.zookeeper.root" "/storm", "supervisor.enable" true, "storm.zookeeper.servers" ["localhost"], "transactional.zookeeper.root" "/transactional", "topology.worker.childop

xumingming

eeper.retry.interval" 1000, "supervisor.slots.ports" (4 5 6), "topology.debug" true, "nimbus.task.launch.secs" 120, "nimbus.supervisor.timeout.secs" 60, "topology.kryo.register" nil, "topology.message.timeout.secs" 3, "task.refresh.poll.secs" 10, "topology.workers" 20, "supervisor.childopts" "-Xmx1024m", "nimbus.thrift.port" 6627, "topology.stats.sample.rate" 0.05, "worker.heartbeat.frequency.secs" 1, "nimbus.task.timeout.secs" 30, "drpc.invocations.port" 3773, "zmq.threads" 1, "storm.zookeeper.retry.times" 5, "topology.state.synchronization.timeout.secs" 60, "supervisor.worker.timeout.secs" 30, "nimbus.file.copy.expiration.secs" 600, "drpc.request.timeout.secs" 600, "storm.local.mode.zmq" false, "ui.port" 8080, "nimbus.childopts" "-Xmx1024m", "topology.ackers" 1, "storm.cluster.mode" "local", "topology.optimize" false, "topology.max.task.parallelism" nil}
35817 [Thread-182] INFO  backtype.storm.daemon.worker  - Worker e2d7e923-be97-48f9-b4e6-e228460a190c for storm test-1-1334943321 on 444bee85-942d-486b-9ef7

xumingming

(defroutes main-routes
  (GET "/" []
       (-> (main-page)
           ui-template))
  (GET "/topology/:id" [id & m]
       (-> (topology-page id (:window m) true)
           ui-template))
  (GET "/topology/:id/component/:component" [id component & m]
       (log-message "REQUEST: " request)
       (-> (component-page id component (:window m) true)

xumingming

Scheduler Interface

I changed List topologies to Map<String, TopologyDetails> topologies, because I think it seems common that people only care about one specific topology, if that topology is missing, the scheduler will just do nothing, so give the developer a map will ease their checking about whether the topology is submitted or not.

public interface Scheduler {
    /**
     * set assignments for all the topologies. The new assignments will be in the returned Clusters.
     */
    public void schedule(Topologies topologies, Cluster cluster);

xumingming

public class RetainSupervisorScheduler implements Scheduler {
    public Cluster schedule(Map<String, TopologyDetails> topologies, Cluster cluster) {
        if (topologies.containsKey("special-topology") && cluster.needSchedule("special-topology") {
            // find the specific supervisor
            SupevisorDetails supervisor = cluster.getSupervisor("specific-supervisor");
            if (supervisor != null) {
                // find out whether this supervisor is occupied
                Map<Integer, ComponentDetails> usedSlots = supervisor.getUsedSlots();
                if (!usedSlots.isEmpty()) {
                    // evict the other topologies's tasks which used these slots