Addressed review comments
Project: http://git-wip-us.apache.org/repos/asf/storm/repo Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/024ff263 Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/024ff263 Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/024ff263 Branch: refs/heads/master Commit: 024ff263800e4032d7b2395fbbf072e0183ebb5b Parents: 664b0ce Author: Robert (Bobby) Evans <[email protected]> Authored: Fri Nov 6 08:16:28 2015 -0600 Committer: Robert (Bobby) Evans <[email protected]> Committed: Fri Nov 6 08:16:28 2015 -0600 ---------------------------------------------------------------------- storm-core/src/clj/backtype/storm/daemon/executor.clj | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/storm/blob/024ff263/storm-core/src/clj/backtype/storm/daemon/executor.clj ---------------------------------------------------------------------- diff --git a/storm-core/src/clj/backtype/storm/daemon/executor.clj b/storm-core/src/clj/backtype/storm/daemon/executor.clj index 2d50770..461f8d6 100644 --- a/storm-core/src/clj/backtype/storm/daemon/executor.clj +++ b/storm-core/src/clj/backtype/storm/daemon/executor.clj @@ -42,7 +42,8 @@ (:require [backtype.storm.daemon.builtin-metrics :as builtin-metrics]) (:require [clojure.set :as set])) -(defn- mk-fields-grouper [^Fields out-fields ^Fields group-fields ^List target-tasks] +(defn- mk-fields-grouper + [^Fields out-fields ^Fields group-fields ^List target-tasks] (let [num-tasks (count target-tasks) task-getter (fn [i] (.get target-tasks i))] (fn [task-id ^List values load] @@ -51,7 +52,8 @@ (mod num-tasks) task-getter)))) -(defn- mk-custom-grouper [^CustomStreamGrouping grouping ^WorkerTopologyContext context ^String component-id ^String stream-id target-tasks] +(defn- mk-custom-grouper + [^CustomStreamGrouping grouping ^WorkerTopologyContext context ^String component-id ^String stream-id target-tasks] (.prepare grouping context (GlobalStreamId. component-id stream-id) target-tasks) (if (instance? LoadAwareCustomStreamGrouping grouping) (fn [task-id ^List values load] @@ -59,7 +61,8 @@ (fn [task-id ^List values load] (.chooseTasks grouping task-id values)))) -(defn mk-shuffle-grouper [^List target-tasks topo-conf ^WorkerTopologyContext context ^String component-id ^String stream-id] +(defn mk-shuffle-grouper + [^List target-tasks topo-conf ^WorkerTopologyContext context ^String component-id ^String stream-id] (if (.get topo-conf TOPOLOGY-DISABLE-LOADAWARE-MESSAGING) (mk-custom-grouper (ShuffleGrouping.) context component-id stream-id target-tasks) (mk-custom-grouper (LoadAwareShuffleGrouping.) context component-id stream-id target-tasks))) @@ -105,7 +108,8 @@ :direct ))) -(defn- outbound-groupings [^WorkerTopologyContext worker-context this-component-id stream-id out-fields component->grouping topo-conf] +(defn- outbound-groupings + [^WorkerTopologyContext worker-context this-component-id stream-id out-fields component->grouping topo-conf] (->> component->grouping (filter-key #(-> worker-context (.getComponentTasks %)
