Repository: storm Updated Branches: refs/heads/master e485e08e0 -> bc54e8ed5
use hb cache instead of fetching via zk Project: http://git-wip-us.apache.org/repos/asf/storm/repo Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/1d973c68 Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/1d973c68 Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/1d973c68 Branch: refs/heads/master Commit: 1d973c6838118e64024505269b82baffd0f41f8d Parents: 94b9608 Author: Derek Dagit <[email protected]> Authored: Wed Jan 14 11:12:30 2015 -0600 Committer: Derek Dagit <[email protected]> Committed: Wed Jan 14 11:12:30 2015 -0600 ---------------------------------------------------------------------- storm-core/src/clj/backtype/storm/daemon/nimbus.clj | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/storm/blob/1d973c68/storm-core/src/clj/backtype/storm/daemon/nimbus.clj ---------------------------------------------------------------------- diff --git a/storm-core/src/clj/backtype/storm/daemon/nimbus.clj b/storm-core/src/clj/backtype/storm/daemon/nimbus.clj index 1fbf7f0..a7dacb8 100644 --- a/storm-core/src/clj/backtype/storm/daemon/nimbus.clj +++ b/storm-core/src/clj/backtype/storm/daemon/nimbus.clj @@ -365,7 +365,8 @@ nimbus-time (>= (time-delta nimbus-time) timeout)) :nimbus-time nimbus-time - :executor-reported-time reported-time})) + :executor-reported-time reported-time + :heartbeat hb})) (defn update-heartbeat-cache [cache executor-beats all-executors timeout] (let [cache (select-keys cache all-executors)] @@ -1265,7 +1266,7 @@ launch-time-secs (if base (:launch-time-secs base) (throw (NotAliveException. (str storm-id)))) task->component (storm-task-info (try-read-storm-topology conf storm-id) topology-conf) assignment (.assignment-info storm-cluster-state storm-id nil) - beats (.executor-beats storm-cluster-state storm-id (:executor->node+port assignment)) + beats (map-val :heartbeat (get @(:heartbeats-cache nimbus) storm-id)) all-components (-> task->component reverse-map keys) errors (->> all-components (map (fn [c] [c (get-errors storm-cluster-state storm-id c)]))
