You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ki...@apache.org on 2015/11/06 18:45:01 UTC
[2/6] storm git commit: STORM-162: Load Aware Shuffle Grouping
STORM-162: Load Aware Shuffle Grouping
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/af64a3a2
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/af64a3a2
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/af64a3a2
Branch: refs/heads/master
Commit: af64a3a2ad951a5753c82d7acffc97518b847dcd
Parents: 7edf520
Author: Bobby Evans <ev...@yahoo-inc.com>
Authored: Tue Feb 10 09:07:21 2015 -0600
Committer: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Committed: Tue Nov 3 12:17:34 2015 -0600
----------------------------------------------------------------------
conf/defaults.yaml | 1 +
.../src/clj/backtype/storm/daemon/executor.clj | 46 ++--
.../src/clj/backtype/storm/daemon/task.clj | 4 +-
.../src/clj/backtype/storm/daemon/worker.clj | 29 +++
.../src/clj/backtype/storm/messaging/local.clj | 34 ++-
storm-core/src/clj/backtype/storm/timer.clj | 20 +-
storm-core/src/jvm/backtype/storm/Config.java | 7 +
.../src/jvm/backtype/storm/grouping/Load.java | 77 ++++++
.../grouping/LoadAwareCustomStreamGrouping.java | 24 ++
.../grouping/LoadAwareShuffleGrouping.java | 76 ++++++
.../backtype/storm/grouping/LoadMapping.java | 64 +++++
.../storm/grouping/ShuffleGrouping.java | 65 ++++++
.../backtype/storm/messaging/IConnection.java | 16 ++
.../backtype/storm/messaging/netty/Client.java | 100 ++++++--
.../storm/messaging/netty/ISaslClient.java | 28 +++
.../storm/messaging/netty/ISaslServer.java | 26 +++
.../backtype/storm/messaging/netty/IServer.java | 26 +++
.../messaging/netty/SaslStormClientHandler.java | 37 +--
.../backtype/storm/messaging/netty/Server.java | 75 ++++--
.../messaging/netty/StormClientHandler.java | 51 +++-
.../netty/StormClientPipelineFactory.java | 11 +-
.../test/clj/backtype/storm/grouping_test.clj | 90 +++++--
.../clj/backtype/storm/integration_test.clj | 4 +-
.../storm/messaging/netty_integration_test.clj | 3 +-
.../storm/messaging/netty_unit_test.clj | 232 ++++++++++++++++---
.../test/clj/backtype/storm/messaging_test.clj | 3 +-
26 files changed, 1005 insertions(+), 144 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/conf/defaults.yaml
----------------------------------------------------------------------
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
index 160c29f..ac1b75b 100644
--- a/conf/defaults.yaml
+++ b/conf/defaults.yaml
@@ -226,6 +226,7 @@ topology.bolts.outgoing.overflow.buffer.enable: false
topology.disruptor.wait.timeout.millis: 1000
topology.disruptor.batch.size: 100
topology.disruptor.batch.timeout.millis: 1
+topology.disable.loadaware: false
# Configs for Resource Aware Scheduler
topology.component.resources.onheap.memory.mb: 128.0
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/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 1bc2db4..f989654 100644
--- a/storm-core/src/clj/backtype/storm/daemon/executor.clj
+++ b/storm-core/src/clj/backtype/storm/daemon/executor.clj
@@ -34,6 +34,7 @@
(:import [backtype.storm.daemon Shutdownable])
(:import [backtype.storm.metric.api IMetric IMetricsConsumer$TaskInfo IMetricsConsumer$DataPoint StateMetric])
(:import [backtype.storm Config Constants])
+ (:import [backtype.storm.grouping LoadAwareCustomStreamGrouping LoadAwareShuffleGrouping LoadMapping ShuffleGrouping])
(:import [java.util.concurrent ConcurrentLinkedQueue])
(:require [backtype.storm [tuple :as tuple] [thrift :as thrift]
[cluster :as cluster] [disruptor :as disruptor] [stats :as stats]])
@@ -44,51 +45,53 @@
(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]
+ (fn [task-id ^List values load]
(-> (.select out-fields group-fields values)
tuple/list-hash-code
(mod num-tasks)
task-getter))))
-(defn- mk-shuffle-grouper [^List target-tasks]
- (let [choices (rotating-random-range target-tasks)]
- (fn [task-id tuple]
- (acquire-random-range-id choices))))
-
(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)
- (fn [task-id ^List values]
- (.chooseTasks grouping task-id values)
- ))
+ (if (instance? LoadAwareCustomStreamGrouping grouping)
+ (fn [task-id ^List values load]
+ (.chooseTasks grouping task-id values load))
+ (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]
+ (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)))
(defn- mk-grouper
"Returns a function that returns a vector of which task indices to send tuple to, or just a single task index."
- [^WorkerTopologyContext context component-id stream-id ^Fields out-fields thrift-grouping ^List target-tasks]
+ [^WorkerTopologyContext context component-id stream-id ^Fields out-fields thrift-grouping ^List target-tasks topo-conf]
(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)
- (fn [task-id tuple]
+ (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))]
(mk-fields-grouper out-fields group-fields target-tasks)
))
:all
- (fn [task-id tuple] target-tasks)
+ (fn [task-id tuple load] target-tasks)
:shuffle
- (mk-shuffle-grouper target-tasks)
+ (mk-shuffle-grouper target-tasks topo-conf context component-id stream-id)
: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))
- (mk-shuffle-grouper target-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
- (fn [task-id tuple]
+ (fn [task-id tuple load]
(let [i (mod (.nextInt random) num-tasks)]
(get target-tasks i)
))
@@ -102,7 +105,7 @@
:direct
)))
-(defn- outbound-groupings [^WorkerTopologyContext worker-context this-component-id stream-id out-fields component->grouping]
+(defn- outbound-groupings [^WorkerTopologyContext worker-context this-component-id stream-id out-fields component->grouping topo-conf]
(->> component->grouping
(filter-key #(-> worker-context
(.getComponentTasks %)
@@ -116,13 +119,13 @@
out-fields
tgrouping
(.getComponentTasks worker-context component)
- )]))
+ topo-conf)]))
(into {})
(HashMap.)))
(defn outbound-components
"Returns map of stream id to component id to grouper"
- [^WorkerTopologyContext worker-context component-id]
+ [^WorkerTopologyContext worker-context component-id topo-conf]
(->> (.getTargets worker-context component-id)
clojurify-structure
(map (fn [[stream-id component->grouping]]
@@ -132,7 +135,8 @@
component-id
stream-id
(.getComponentOutputFields worker-context component-id stream-id)
- component->grouping)]))
+ component->grouping
+ topo-conf)]))
(into {})
(HashMap.)))
@@ -240,7 +244,7 @@
:stats (mk-executor-stats <> (sampling-rate storm-conf))
:interval->task->metric-registry (HashMap.)
:task->component (:task->component worker)
- :stream->component->grouper (outbound-components worker-context component-id)
+ :stream->component->grouper (outbound-components worker-context component-id storm-conf)
:report-error (throttled-report-error-fn <>)
:report-error-and-die (fn [error]
((:report-error <>) error)
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/clj/backtype/storm/daemon/task.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/daemon/task.clj b/storm-core/src/clj/backtype/storm/daemon/task.clj
index c48fb03..7133fdf 100644
--- a/storm-core/src/clj/backtype/storm/daemon/task.clj
+++ b/storm-core/src/clj/backtype/storm/daemon/task.clj
@@ -18,6 +18,7 @@
(:use [backtype.storm config util log])
(:import [backtype.storm.hooks ITaskHook])
(:import [backtype.storm.tuple Tuple TupleImpl])
+ (:import [backtype.storm.grouping LoadMapping])
(:import [backtype.storm.generated SpoutSpec Bolt StateSpoutSpec StormTopology])
(:import [backtype.storm.hooks.info SpoutAckInfo SpoutFailInfo
EmitInfo BoltFailInfo BoltAckInfo])
@@ -119,6 +120,7 @@
(defn mk-tasks-fn [task-data]
(let [task-id (:task-id task-data)
executor-data (:executor-data task-data)
+ ^LoadMapping load-mapping (:load-mapping (:worker executor-data))
component-id (:component-id executor-data)
^WorkerTopologyContext worker-context (:worker-context executor-data)
storm-conf (:storm-conf executor-data)
@@ -152,7 +154,7 @@
(when (= :direct grouper)
;; TODO: this is wrong, need to check how the stream was declared
(throw (IllegalArgumentException. "Cannot do regular emit to direct stream")))
- (let [comp-tasks (grouper task-id values)]
+ (let [comp-tasks (grouper task-id values load-mapping)]
(if (or (sequential? comp-tasks) (instance? Collection comp-tasks))
(.addAll out-tasks comp-tasks)
(.add out-tasks comp-tasks)
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/clj/backtype/storm/daemon/worker.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/daemon/worker.clj b/storm-core/src/clj/backtype/storm/daemon/worker.clj
index f304bdd..453801d 100644
--- a/storm-core/src/clj/backtype/storm/daemon/worker.clj
+++ b/storm-core/src/clj/backtype/storm/daemon/worker.clj
@@ -25,6 +25,7 @@
(:import [java.util.concurrent Executors])
(:import [java.util ArrayList HashMap])
(:import [backtype.storm.utils Utils TransferDrainer ThriftTopologyUtils WorkerBackpressureThread DisruptorQueue])
+ (:import [backtype.storm.grouping LoadMapping])
(:import [backtype.storm.messaging TransportFactory])
(:import [backtype.storm.messaging TaskMessage IContext IConnection ConnectionWithStatus ConnectionWithStatus$Status])
(:import [backtype.storm.daemon Shutdownable])
@@ -269,6 +270,7 @@
:topology topology
:system-topology (system-topology! storm-conf topology)
:heartbeat-timer (mk-halting-timer "heartbeat-timer")
+ :refresh-load-timer (mk-halting-timer "refresh-load-timer")
:refresh-connections-timer (mk-halting-timer "refresh-connections-timer")
:refresh-credentials-timer (mk-halting-timer "refresh-credentials-timer")
:reset-log-levels-timer (mk-halting-timer "reset-log-levels-timer")
@@ -295,6 +297,7 @@
:transfer-local-fn (mk-transfer-local-fn <>)
:receiver-thread-count (get storm-conf WORKER-RECEIVER-THREAD-COUNT)
:transfer-fn (mk-transfer-fn <>)
+ :load-mapping (LoadMapping.)
:assignment-versions assignment-versions
:backpressure (atom false) ;; whether this worker is going slow
:transfer-backpressure (atom false) ;; if the transfer queue is backed-up
@@ -310,6 +313,26 @@
[node (Integer/valueOf port-str)]
))
+(defn mk-refresh-load [worker]
+ (let [local-tasks (set (:task-ids worker))
+ remote-tasks (set/difference (worker-outbound-tasks worker) local-tasks)
+ short-executor-receive-queue-map (:short-executor-receive-queue-map worker)
+ next-update (atom 0)]
+ (fn this
+ ([]
+ (let [^LoadMapping load-mapping (:load-mapping worker)
+ local-pop (map-val (fn [queue]
+ (let [q-metrics (.getMetrics queue)]
+ (/ (double (.population q-metrics)) (.capacity q-metrics))))
+ short-executor-receive-queue-map)
+ remote-load (reduce merge (for [[np conn] @(:cached-node+port->socket worker)] (into {} (.getLoad conn remote-tasks))))
+ now (System/currentTimeMillis)]
+ (.setLocal load-mapping local-pop)
+ (.setRemote load-mapping remote-load)
+ (when (> now @next-update)
+ (.sendLoadMetrics (:receiver worker) local-pop)
+ (reset! next-update (+ 5000 now))))))))
+
(defn mk-refresh-connections [worker]
(let [outbound-tasks (worker-outbound-tasks worker)
conf (:conf worker)
@@ -572,6 +595,7 @@
receive-thread-shutdown (launch-receive-thread worker)
refresh-connections (mk-refresh-connections worker)
+ refresh-load (mk-refresh-load worker)
_ (refresh-connections nil)
@@ -633,6 +657,7 @@
(cancel-timer (:refresh-active-timer worker))
(cancel-timer (:executor-heartbeat-timer worker))
(cancel-timer (:user-timer worker))
+ (cancel-timer (:refresh-load-timer worker))
(close-resources worker)
@@ -653,6 +678,7 @@
(and
(timer-waiting? (:heartbeat-timer worker))
(timer-waiting? (:refresh-connections-timer worker))
+ (timer-waiting? (:refresh-load-timer worker))
(timer-waiting? (:refresh-credentials-timer worker))
(timer-waiting? (:refresh-active-timer worker))
(timer-waiting? (:executor-heartbeat-timer worker))
@@ -689,6 +715,9 @@
(check-credentials-changed)
(if ((:storm-conf worker) TOPOLOGY-BACKPRESSURE-ENABLE)
(check-throttle-changed))))
+ ;; The jitter allows the clients to get the data at different times, and avoids thundering herd
+ (when-not (.get conf TOPOLOGY-DISABLE-LOADAWARE-MESSAGING)
+ (schedule-recurring-with-jitter (:refresh-load-timer worker) 0 1 500 refresh-load))
(schedule-recurring (:refresh-connections-timer worker) 0 (conf TASK-REFRESH-POLL-SECS) refresh-connections)
(schedule-recurring (:reset-log-levels-timer worker) 0 (conf WORKER-LOG-LEVEL-RESET-POLL-SECS) (fn [] (reset-log-levels latest-log-config)))
(schedule-recurring (:refresh-active-timer worker) 0 (conf TASK-REFRESH-POLL-SECS) (partial refresh-storm-active worker))
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/clj/backtype/storm/messaging/local.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/messaging/local.clj b/storm-core/src/clj/backtype/storm/messaging/local.clj
index 4aa67ab..60a6bd2 100644
--- a/storm-core/src/clj/backtype/storm/messaging/local.clj
+++ b/storm-core/src/clj/backtype/storm/messaging/local.clj
@@ -15,13 +15,18 @@
;; limitations under the License.
(ns backtype.storm.messaging.local
(:refer-clojure :exclude [send])
- (:use [backtype.storm log])
+ (:use [backtype.storm log util])
(:import [backtype.storm.messaging IContext IConnection TaskMessage])
+ (:import [backtype.storm.grouping Load])
(:import [java.util.concurrent LinkedBlockingQueue])
- (:import [java.util Map Iterator])
+ (:import [java.util Map Iterator Collection])
(:import [java.util Iterator ArrayList])
(:gen-class))
+(defn update-load! [cached-task->load lock task->load]
+ (locking lock
+ (swap! cached-task->load merge task->load)))
+
(defn add-queue! [queues-map lock storm-id port]
(let [id (str storm-id "-" port)]
(locking lock
@@ -29,7 +34,7 @@
(swap! queues-map assoc id (LinkedBlockingQueue.))))
(@queues-map id)))
-(deftype LocalConnection [storm-id port queues-map lock queue]
+(deftype LocalConnection [storm-id port queues-map lock queue task->load]
IConnection
(^Iterator recv [this ^int flags ^int clientId]
(when-not queue
@@ -50,24 +55,35 @@
(while (.hasNext iter)
(.put send-queue (.next iter)))
))
- (^void close [this]
- ))
+ (^void sendLoadMetrics [this ^Map taskToLoad]
+ (update-load! task->load lock taskToLoad))
+ (^Map getLoad [this ^Collection tasks]
+ (locking lock
+ (into {}
+ (for [task tasks
+ :let [load (.get @task->load task)]
+ :when (not-nil? load)]
+ ;; for now we are ignoring the connection load locally
+ [task (Load. true load 0.0)]))))
+ (^void close [this]))
(deftype LocalContext [^{:unsynchronized-mutable true} queues-map
- ^{:unsynchronized-mutable true} lock]
+ ^{:unsynchronized-mutable true} lock
+ ^{:unsynchronized-mutable true} task->load]
IContext
(^void prepare [this ^Map storm-conf]
(set! queues-map (atom {}))
+ (set! task->load (atom {}))
(set! lock (Object.)))
(^IConnection bind [this ^String storm-id ^int port]
- (LocalConnection. storm-id port queues-map lock (add-queue! queues-map lock storm-id port)))
+ (LocalConnection. storm-id port queues-map lock (add-queue! queues-map lock storm-id port) task->load))
(^IConnection connect [this ^String storm-id ^String host ^int port]
- (LocalConnection. storm-id port queues-map lock nil))
+ (LocalConnection. storm-id port queues-map lock nil task->load))
(^void term [this]
))
(defn mk-context []
- (let [context (LocalContext. nil nil)]
+ (let [context (LocalContext. nil nil nil)]
(.prepare ^IContext context nil)
context))
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/clj/backtype/storm/timer.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/timer.clj b/storm-core/src/clj/backtype/storm/timer.clj
index 2c76ce2..b5f73f7 100644
--- a/storm-core/src/clj/backtype/storm/timer.clj
+++ b/storm-core/src/clj/backtype/storm/timer.clj
@@ -16,7 +16,7 @@
(ns backtype.storm.timer
(:import [backtype.storm.utils Time])
- (:import [java.util PriorityQueue Comparator])
+ (:import [java.util PriorityQueue Comparator Random])
(:import [java.util.concurrent Semaphore])
(:use [backtype.storm util log]))
@@ -79,6 +79,7 @@
:queue queue
:active active
:lock lock
+ :random (Random.)
:cancel-notifier notifier}))
(defn- check-active!
@@ -87,12 +88,14 @@
(throw (IllegalStateException. "Timer is not active"))))
(defnk schedule
- [timer delay-secs afn :check-active true]
+ [timer delay-secs afn :check-active true :jitter-ms 0]
(when check-active (check-active! timer))
(let [id (uuid)
- ^PriorityQueue queue (:queue timer)]
+ ^PriorityQueue queue (:queue timer)
+ end-time-ms (+ (current-time-millis) (secs-to-millis-long delay-secs))
+ end-time-ms (if (< 0 jitter-ms) (+ (.nextInt (:random timer) jitter-ms) end-time-ms) end-time-ms)]
(locking (:lock timer)
- (.add queue [(+ (current-time-millis) (secs-to-millis-long delay-secs)) afn id]))))
+ (.add queue [end-time-ms afn id]))))
(defn schedule-recurring
[timer delay-secs recur-secs afn]
@@ -103,6 +106,15 @@
; This avoids a race condition with cancel-timer.
(schedule timer recur-secs this :check-active false))))
+(defn schedule-recurring-with-jitter
+ [timer delay-secs recur-secs jitter-ms afn]
+ (schedule timer
+ delay-secs
+ (fn this []
+ (afn)
+ ; This avoids a race condition with cancel-timer.
+ (schedule timer recur-secs this :check-active false :jitter-ms jitter-ms))))
+
(defn cancel-timer
[timer]
(check-active! timer)
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/jvm/backtype/storm/Config.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/Config.java b/storm-core/src/jvm/backtype/storm/Config.java
index 73e41cb..eab1575 100644
--- a/storm-core/src/jvm/backtype/storm/Config.java
+++ b/storm-core/src/jvm/backtype/storm/Config.java
@@ -239,6 +239,13 @@ public class Config extends HashMap<String, Object> {
public static final String TOPOLOGY_TUPLE_SERIALIZER = "topology.tuple.serializer";
/**
+ * Disable load aware grouping support.
+ */
+ @isBoolean
+ @NotNull
+ public static final String TOPOLOGY_DISABLE_LOADAWARE_MESSAGING = "topology.disable.loadaware.messaging";
+
+ /**
* Try to serialize all tuples, even for local transfers. This should only be used
* for testing, as a sanity check that all of your tuples are setup properly.
*/
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/jvm/backtype/storm/grouping/Load.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/grouping/Load.java b/storm-core/src/jvm/backtype/storm/grouping/Load.java
new file mode 100644
index 0000000..a10c791
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/grouping/Load.java
@@ -0,0 +1,77 @@
+/**
+ * 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 backtype.storm.grouping;
+
+import java.io.Serializable;
+
+/**
+ * Represents the load that a Bolt is currently under to help in
+ * deciding where to route a tuple, to help balance the load.
+ */
+public class Load {
+ private boolean hasMetrics = false;
+ private double boltLoad = 0.0; //0 no load to 1 fully loaded
+ private double connectionLoad = 0.0; //0 no load to 1 fully loaded
+
+ /**
+ * Create a new load
+ * @param hasMetrics have metrics been reported yet?
+ * @param boltLoad the load as reported by the bolt 0.0 no load 1.0 fully loaded
+ * @param connectionLoad the load as reported by the connection to the bolt 0.0 no load 1.0 fully loaded.
+ */
+ public Load(boolean hasMetrics, double boltLoad, double connectionLoad) {
+ this.hasMetrics = hasMetrics;
+ this.boltLoad = boltLoad;
+ this.connectionLoad = connectionLoad;
+ }
+
+ /**
+ * @return true if metrics have been reported so far.
+ */
+ public boolean hasMetrics() {
+ return hasMetrics;
+ }
+
+ /**
+ * @return the load as reported by the bolt.
+ */
+ public double getBoltLoad() {
+ return boltLoad;
+ }
+
+ /**
+ * @return the load as reported by the connection
+ */
+ public double getConnectionLoad() {
+ return connectionLoad;
+ }
+
+ /**
+ * @return the load that is a combination of sub loads.
+ */
+ public double getLoad() {
+ if (!hasMetrics) {
+ return 1.0;
+ }
+ return connectionLoad > boltLoad ? connectionLoad : boltLoad;
+ }
+
+ public String toString() {
+ return "[:load "+boltLoad+" "+connectionLoad+"]";
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/jvm/backtype/storm/grouping/LoadAwareCustomStreamGrouping.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/grouping/LoadAwareCustomStreamGrouping.java b/storm-core/src/jvm/backtype/storm/grouping/LoadAwareCustomStreamGrouping.java
new file mode 100644
index 0000000..f39273f
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/grouping/LoadAwareCustomStreamGrouping.java
@@ -0,0 +1,24 @@
+/**
+ * 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 backtype.storm.grouping;
+
+import java.util.List;
+
+public interface LoadAwareCustomStreamGrouping extends CustomStreamGrouping {
+ List<Integer> chooseTasks(int taskId, List<Object> values, LoadMapping load);
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/jvm/backtype/storm/grouping/LoadAwareShuffleGrouping.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/grouping/LoadAwareShuffleGrouping.java b/storm-core/src/jvm/backtype/storm/grouping/LoadAwareShuffleGrouping.java
new file mode 100644
index 0000000..a19b585
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/grouping/LoadAwareShuffleGrouping.java
@@ -0,0 +1,76 @@
+/**
+ * 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 backtype.storm.grouping;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Random;
+
+import backtype.storm.generated.GlobalStreamId;
+import backtype.storm.task.WorkerTopologyContext;
+
+public class LoadAwareShuffleGrouping implements LoadAwareCustomStreamGrouping, Serializable {
+ private Random random;
+ private List<Integer>[] rets;
+ private int[] targets;
+ private int[] loads;
+ private int total;
+ private long lastUpdate = 0;
+
+ @Override
+ public void prepare(WorkerTopologyContext context, GlobalStreamId stream, List<Integer> targetTasks) {
+ random = new Random();
+ rets = (List<Integer>[])new List<?>[targetTasks.size()];
+ targets = new int[targetTasks.size()];
+ for (int i = 0; i < targets.length; i++) {
+ rets[i] = Arrays.asList(targetTasks.get(i));
+ targets[i] = targetTasks.get(i);
+ }
+ loads = new int[targets.length];
+ }
+
+ @Override
+ public List<Integer> chooseTasks(int taskId, List<Object> values) {
+ throw new RuntimeException("NOT IMPLEMENTED");
+ }
+
+ @Override
+ public List<Integer> chooseTasks(int taskId, List<Object> values, LoadMapping load) {
+ if ((lastUpdate + 1000) < System.currentTimeMillis()) {
+ int local_total = 0;
+ for (int i = 0; i < targets.length; i++) {
+ int val = (int)(101 - (load.get(targets[i]) * 100));
+ loads[i] = val;
+ local_total += val;
+ }
+ total = local_total;
+ lastUpdate = System.currentTimeMillis();
+ }
+ int selected = random.nextInt(total);
+ int sum = 0;
+ for (int i = 0; i < targets.length; i++) {
+ sum += loads[i];
+ if (selected < sum) {
+ return rets[i];
+ }
+ }
+ return rets[rets.length-1];
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/jvm/backtype/storm/grouping/LoadMapping.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/grouping/LoadMapping.java b/storm-core/src/jvm/backtype/storm/grouping/LoadMapping.java
new file mode 100644
index 0000000..abd8983
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/grouping/LoadMapping.java
@@ -0,0 +1,64 @@
+/**
+ * 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 backtype.storm.grouping;
+
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.Map;
+import java.util.HashMap;
+
+/**
+ * Holds a list of the current loads
+ */
+public class LoadMapping {
+ private static final Load NOT_CONNECTED = new Load(false, 1.0, 1.0);
+ private final AtomicReference<Map<Integer,Load>> _local = new AtomicReference<Map<Integer,Load>>(new HashMap<Integer,Load>());
+ private final AtomicReference<Map<Integer,Load>> _remote = new AtomicReference<Map<Integer,Load>>(new HashMap<Integer,Load>());
+
+ public void setLocal(Map<Integer, Double> local) {
+ Map<Integer, Load> newLocal = new HashMap<Integer, Load>();
+ if (local != null) {
+ for (Map.Entry<Integer, Double> entry: local.entrySet()) {
+ newLocal.put(entry.getKey(), new Load(true, entry.getValue(), 0.0));
+ }
+ }
+ _local.set(newLocal);
+ }
+
+ public void setRemote(Map<Integer, Load> remote) {
+ if (remote != null) {
+ _remote.set(new HashMap<Integer, Load>(remote));
+ } else {
+ _remote.set(new HashMap<Integer, Load>());
+ }
+ }
+
+ public Load getLoad(int task) {
+ Load ret = _local.get().get(task);
+ if (ret == null) {
+ ret = _remote.get().get(task);
+ }
+ if (ret == null) {
+ ret = NOT_CONNECTED;
+ }
+ return ret;
+ }
+
+ public double get(int task) {
+ return getLoad(task).getLoad();
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/jvm/backtype/storm/grouping/ShuffleGrouping.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/grouping/ShuffleGrouping.java b/storm-core/src/jvm/backtype/storm/grouping/ShuffleGrouping.java
new file mode 100644
index 0000000..01755bb
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/grouping/ShuffleGrouping.java
@@ -0,0 +1,65 @@
+/**
+ * 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 backtype.storm.grouping;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import backtype.storm.generated.GlobalStreamId;
+import backtype.storm.task.WorkerTopologyContext;
+
+public class ShuffleGrouping implements CustomStreamGrouping, Serializable {
+ private Random random;
+ private ArrayList<List<Integer>> choices;
+ private AtomicInteger current;
+
+ @Override
+ public void prepare(WorkerTopologyContext context, GlobalStreamId stream, List<Integer> targetTasks) {
+ random = new Random();
+ choices = new ArrayList<List<Integer>>(targetTasks.size());
+ for (Integer i: targetTasks) {
+ choices.add(Arrays.asList(i));
+ }
+ Collections.shuffle(choices, random);
+ current = new AtomicInteger(0);
+ }
+
+ @Override
+ public List<Integer> chooseTasks(int taskId, List<Object> values) {
+ int rightNow;
+ int size = choices.size();
+ while (true) {
+ rightNow = current.incrementAndGet();
+ if (rightNow < size) {
+ return choices.get(rightNow);
+ } else if (rightNow == size) {
+ current.set(0);
+ //This should be thread safe so long as ArrayList does not have any internal state that can be messed up by multi-treaded access.
+ Collections.shuffle(choices, random);
+ return choices.get(0);
+ }
+ //race condition with another thread, and we lost
+ // try again
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/jvm/backtype/storm/messaging/IConnection.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/messaging/IConnection.java b/storm-core/src/jvm/backtype/storm/messaging/IConnection.java
index ead4935..88f0c0a 100644
--- a/storm-core/src/jvm/backtype/storm/messaging/IConnection.java
+++ b/storm-core/src/jvm/backtype/storm/messaging/IConnection.java
@@ -17,7 +17,10 @@
*/
package backtype.storm.messaging;
+import backtype.storm.grouping.Load;
+import java.util.Collection;
import java.util.Iterator;
+import java.util.Map;
public interface IConnection {
@@ -27,6 +30,12 @@ public interface IConnection {
* @return
*/
public Iterator<TaskMessage> recv(int flags, int clientId);
+
+ /**
+ * Send load metrics to all downstream connections.
+ * @param taskToLoad a map from the task id to the load for that task.
+ */
+ public void sendLoadMetrics(Map<Integer, Double> taskToLoad);
/**
* send a message with taskId and payload
@@ -43,6 +52,13 @@ public interface IConnection {
public void send(Iterator<TaskMessage> msgs);
/**
+ * Get the current load for the given tasks
+ * @param tasks the tasks to look for.
+ * @return a Load for each of the tasks it knows about.
+ */
+ public Map<Integer, Load> getLoad(Collection<Integer> tasks);
+
+ /**
* close this connection
*/
public void close();
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/jvm/backtype/storm/messaging/netty/Client.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/Client.java b/storm-core/src/jvm/backtype/storm/messaging/netty/Client.java
index 2149c0d..b9151f3 100644
--- a/storm-core/src/jvm/backtype/storm/messaging/netty/Client.java
+++ b/storm-core/src/jvm/backtype/storm/messaging/netty/Client.java
@@ -17,7 +17,19 @@
*/
package backtype.storm.messaging.netty;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.Iterator;
+import java.util.Collection;
+import java.util.Map;
+import java.util.HashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.lang.InterruptedException;
+
import backtype.storm.Config;
+import backtype.storm.grouping.Load;
import backtype.storm.messaging.ConnectionWithStatus;
import backtype.storm.messaging.TaskMessage;
import backtype.storm.metric.api.IStatefulObject;
@@ -34,17 +46,11 @@ import org.jboss.netty.util.TimerTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.net.InetSocketAddress;
-import java.net.SocketAddress;
+
import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Iterator;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicReference;
import static com.google.common.base.Preconditions.checkState;
@@ -60,7 +66,7 @@ import static com.google.common.base.Preconditions.checkState;
* - Note: The current implementation drops any messages that are being enqueued for sending if the connection to
* the remote destination is currently unavailable.
*/
-public class Client extends ConnectionWithStatus implements IStatefulObject {
+public class Client extends ConnectionWithStatus implements IStatefulObject, ISaslClient {
private static final long PENDING_MESSAGES_FLUSH_TIMEOUT_MS = 600000L;
private static final long PENDING_MESSAGES_FLUSH_INTERVAL_MS = 1000L;
@@ -73,6 +79,7 @@ public class Client extends ConnectionWithStatus implements IStatefulObject {
private final ClientBootstrap bootstrap;
private final InetSocketAddress dstAddress;
protected final String dstAddressPrefixedName;
+ private volatile Map<Integer, Double> serverLoad = null;
/**
* The channel used for all write operations from this client to the remote destination.
@@ -104,6 +111,10 @@ public class Client extends ConnectionWithStatus implements IStatefulObject {
*/
private final AtomicLong pendingMessages = new AtomicLong(0);
+ /**
+ * Whether the SASL channel is ready.
+ */
+ private final AtomicBoolean saslChannelReady = new AtomicBoolean(false);
/**
* This flag is set to true if and only if a client instance is being closed.
@@ -125,6 +136,8 @@ public class Client extends ConnectionWithStatus implements IStatefulObject {
this.scheduler = scheduler;
this.context = context;
int bufferSize = Utils.getInt(stormConf.get(Config.STORM_MESSAGING_NETTY_BUFFER_SIZE));
+ // if SASL authentication is disabled, saslChannelReady is initialized as true; otherwise false
+ saslChannelReady.set(!Utils.getBoolean(stormConf.get(Config.STORM_MESSAGING_NETTY_AUTHENTICATION), false));
LOG.info("creating Netty Client, connecting to {}:{}, bufferSize: {}", host, port, bufferSize);
int messageBatchSize = Utils.getInt(stormConf.get(Config.STORM_NETTY_MESSAGE_BATCH_SIZE), 262144);
@@ -134,19 +147,19 @@ public class Client extends ConnectionWithStatus implements IStatefulObject {
retryPolicy = new StormBoundedExponentialBackoffRetry(minWaitMs, maxWaitMs, maxReconnectionAttempts);
// Initiate connection to remote destination
- bootstrap = createClientBootstrap(factory, bufferSize);
+ bootstrap = createClientBootstrap(factory, bufferSize, stormConf);
dstAddress = new InetSocketAddress(host, port);
dstAddressPrefixedName = prefixedName(dstAddress);
scheduleConnect(NO_DELAY_MS);
batcher = new MessageBuffer(messageBatchSize);
}
- private ClientBootstrap createClientBootstrap(ChannelFactory factory, int bufferSize) {
+ private ClientBootstrap createClientBootstrap(ChannelFactory factory, int bufferSize, Map stormConf) {
ClientBootstrap bootstrap = new ClientBootstrap(factory);
bootstrap.setOption("tcpNoDelay", true);
bootstrap.setOption("sendBufferSize", bufferSize);
bootstrap.setOption("keepAlive", true);
- bootstrap.setPipelineFactory(new StormClientPipelineFactory(this));
+ bootstrap.setPipelineFactory(new StormClientPipelineFactory(this, stormConf));
return bootstrap;
}
@@ -158,7 +171,7 @@ public class Client extends ConnectionWithStatus implements IStatefulObject {
}
/**
- * We will retry connection with exponential back-off policy
+ * Enqueue a task message to be sent to server
*/
private void scheduleConnect(long delayMs) {
scheduler.newTimeout(new Connect(dstAddress), delayMs, TimeUnit.MILLISECONDS);
@@ -190,7 +203,11 @@ public class Client extends ConnectionWithStatus implements IStatefulObject {
} else if (!connectionEstablished(channelRef.get())) {
return Status.Connecting;
} else {
- return Status.Ready;
+ if (saslChannelReady.get()) {
+ return Status.Ready;
+ } else {
+ return Status.Connecting; // need to wait until sasl channel is also ready
+ }
}
}
@@ -205,6 +222,11 @@ public class Client extends ConnectionWithStatus implements IStatefulObject {
}
@Override
+ public void sendLoadMetrics(Map<Integer, Double> taskToLoad) {
+ throw new RuntimeException("Client connection should not send load metrics");
+ }
+
+ @Override
public void send(int taskId, byte[] payload) {
TaskMessage msg = new TaskMessage(taskId, payload);
List<TaskMessage> wrapper = new ArrayList<TaskMessage>(1);
@@ -283,11 +305,14 @@ public class Client extends ConnectionWithStatus implements IStatefulObject {
}
}
+ public InetSocketAddress getDstAddress() {
+ return dstAddress;
+ }
+
private boolean hasMessages(Iterator<TaskMessage> msgs) {
return msgs != null && msgs.hasNext();
}
-
private void dropMessages(Iterator<TaskMessage> msgs) {
// We consume the iterator by traversing and thus "emptying" it.
int msgCount = iteratorSize(msgs);
@@ -383,15 +408,13 @@ public class Client extends ConnectionWithStatus implements IStatefulObject {
break;
}
Thread.sleep(PENDING_MESSAGES_FLUSH_INTERVAL_MS);
- }
- catch (InterruptedException e) {
+ } catch (InterruptedException e) {
break;
}
}
}
-
private void closeChannel() {
Channel channel = channelRef.get();
if (channel != null) {
@@ -400,9 +423,29 @@ public class Client extends ConnectionWithStatus implements IStatefulObject {
}
}
+ void setLoadMetrics(Map<Integer, Double> taskToLoad) {
+ this.serverLoad = taskToLoad;
+ }
+
+ @Override
+ public Map<Integer, Load> getLoad(Collection<Integer> tasks) {
+ Map<Integer, Double> loadCache = serverLoad;
+ Map<Integer, Load> ret = new HashMap<Integer, Load>();
+ if (loadCache != null) {
+ double clientLoad = Math.min(pendingMessages.get(), 1024)/1024.0;
+ for (Integer task : tasks) {
+ Double found = loadCache.get(task);
+ if (found != null) {
+ ret.put(task, new Load(true, found, clientLoad));
+ }
+ }
+ }
+ return ret;
+ }
+
@Override
public Object getState() {
- LOG.info("Getting metrics for client connection to {}", dstAddressPrefixedName);
+ LOG.debug("Getting metrics for client connection to {}", dstAddressPrefixedName);
HashMap<String, Object> ret = new HashMap<String, Object>();
ret.put("reconnects", totalConnectionAttempts.getAndSet(0));
ret.put("sent", messagesSent.getAndSet(0));
@@ -416,10 +459,28 @@ public class Client extends ConnectionWithStatus implements IStatefulObject {
return ret;
}
- public Map getStormConf() {
+ public Map getConfig() {
return stormConf;
}
+ /** ISaslClient interface **/
+ public void channelConnected(Channel channel) {
+// setChannel(channel);
+ }
+
+ public void channelReady() {
+ saslChannelReady.set(true);
+ }
+
+ public String name() {
+ return (String)stormConf.get(Config.TOPOLOGY_NAME);
+ }
+
+ public String secretKey() {
+ return SaslUtils.getSecretKey(stormConf);
+ }
+ /** end **/
+
private String srcAddressName() {
String name = null;
Channel channel = channelRef.get();
@@ -513,5 +574,4 @@ public class Client extends ConnectionWithStatus implements IStatefulObject {
}
}
}
-
}
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/jvm/backtype/storm/messaging/netty/ISaslClient.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/ISaslClient.java b/storm-core/src/jvm/backtype/storm/messaging/netty/ISaslClient.java
new file mode 100644
index 0000000..57dcfe8
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/messaging/netty/ISaslClient.java
@@ -0,0 +1,28 @@
+/**
+ * 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 backtype.storm.messaging.netty;
+
+import org.jboss.netty.channel.Channel;
+import backtype.storm.Config;
+
+public interface ISaslClient {
+ void channelConnected(Channel channel);
+ void channelReady();
+ String name();
+ String secretKey();
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/jvm/backtype/storm/messaging/netty/ISaslServer.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/ISaslServer.java b/storm-core/src/jvm/backtype/storm/messaging/netty/ISaslServer.java
new file mode 100644
index 0000000..4203dcc
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/messaging/netty/ISaslServer.java
@@ -0,0 +1,26 @@
+/**
+ * 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 backtype.storm.messaging.netty;
+
+import org.jboss.netty.channel.Channel;
+
+public interface ISaslServer extends IServer {
+ String name();
+ String secretKey();
+ void authenticated(Channel c);
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/jvm/backtype/storm/messaging/netty/IServer.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/IServer.java b/storm-core/src/jvm/backtype/storm/messaging/netty/IServer.java
new file mode 100644
index 0000000..d046492
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/messaging/netty/IServer.java
@@ -0,0 +1,26 @@
+/**
+ * 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 backtype.storm.messaging.netty;
+
+import org.jboss.netty.channel.Channel;
+
+public interface IServer {
+ void channelConnected(Channel c);
+ void received(Object message, String remote, Channel channel) throws InterruptedException;
+ void closeChannel(Channel c);
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/jvm/backtype/storm/messaging/netty/SaslStormClientHandler.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/SaslStormClientHandler.java b/storm-core/src/jvm/backtype/storm/messaging/netty/SaslStormClientHandler.java
index 12b466c..2a5ae99 100644
--- a/storm-core/src/jvm/backtype/storm/messaging/netty/SaslStormClientHandler.java
+++ b/storm-core/src/jvm/backtype/storm/messaging/netty/SaslStormClientHandler.java
@@ -28,19 +28,18 @@ import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import backtype.storm.Config;
-
public class SaslStormClientHandler extends SimpleChannelUpstreamHandler {
private static final Logger LOG = LoggerFactory
.getLogger(SaslStormClientHandler.class);
- private Client client;
+
+ private ISaslClient client;
long start_time;
/** Used for client or server's token to send or receive from each other. */
private byte[] token;
- private String topologyName;
+ private String name;
- public SaslStormClientHandler(Client client) throws IOException {
+ public SaslStormClientHandler(ISaslClient client) throws IOException {
this.client = client;
start_time = System.currentTimeMillis();
getSASLCredentials();
@@ -51,9 +50,7 @@ public class SaslStormClientHandler extends SimpleChannelUpstreamHandler {
ChannelStateEvent event) {
// register the newly established channel
Channel channel = ctx.getChannel();
-
- LOG.info("Connection established from " + channel.getLocalAddress()
- + " to " + channel.getRemoteAddress());
+ client.channelConnected(channel);
try {
SaslNettyClient saslNettyClient = SaslNettyClientState.getSaslNettyClient
@@ -62,10 +59,11 @@ public class SaslStormClientHandler extends SimpleChannelUpstreamHandler {
if (saslNettyClient == null) {
LOG.debug("Creating saslNettyClient now " + "for channel: "
+ channel);
- saslNettyClient = new SaslNettyClient(topologyName, token);
+ saslNettyClient = new SaslNettyClient(name, token);
SaslNettyClientState.getSaslNettyClient.set(channel,
saslNettyClient);
}
+ LOG.debug("Sending SASL_TOKEN_MESSAGE_REQUEST");
channel.write(ControlMessage.SASL_TOKEN_MESSAGE_REQUEST);
} catch (Exception e) {
LOG.error("Failed to authenticate with server " + "due to error: ",
@@ -96,7 +94,7 @@ public class SaslStormClientHandler extends SimpleChannelUpstreamHandler {
ControlMessage msg = (ControlMessage) event.getMessage();
if (msg == ControlMessage.SASL_COMPLETE_REQUEST) {
LOG.debug("Server has sent us the SaslComplete "
- + "message. Allowing normal work to proceed.");
+ + "message. Allowing normal work to proceed.");
if (!saslNettyClient.isComplete()) {
LOG.error("Server returned a Sasl-complete message, "
@@ -106,6 +104,8 @@ public class SaslStormClientHandler extends SimpleChannelUpstreamHandler {
+ "we can tell, we are not authenticated yet.");
}
ctx.getPipeline().remove(this);
+ this.client.channelReady();
+
// We call fireMessageReceived since the client is allowed to
// perform this request. The client's request will now proceed
// to the next pipeline component namely StormClientHandler.
@@ -116,7 +116,7 @@ public class SaslStormClientHandler extends SimpleChannelUpstreamHandler {
SaslMessageToken saslTokenMessage = (SaslMessageToken) event
.getMessage();
LOG.debug("Responding to server's token of length: "
- + saslTokenMessage.getSaslToken().length);
+ + saslTokenMessage.getSaslToken().length);
// Generate SASL response (but we only actually send the response if
// it's non-null.
@@ -127,17 +127,18 @@ public class SaslStormClientHandler extends SimpleChannelUpstreamHandler {
// (if not, warn), and return without sending a response back to the
// server.
LOG.debug("Response to server is null: "
- + "authentication should now be complete.");
+ + "authentication should now be complete.");
if (!saslNettyClient.isComplete()) {
LOG.warn("Generated a null response, "
+ "but authentication is not complete.");
throw new Exception("Server reponse is null, but as far as "
+ "we can tell, we are not authenticated yet.");
}
+ this.client.channelReady();
return;
} else {
LOG.debug("Response to server token has length:"
- + responseToServer.length);
+ + responseToServer.length);
}
// Construct a message containing the SASL response and send it to the
// server.
@@ -146,12 +147,14 @@ public class SaslStormClientHandler extends SimpleChannelUpstreamHandler {
}
private void getSASLCredentials() throws IOException {
- topologyName = (String) this.client.getStormConf().get(Config.TOPOLOGY_NAME);
- String secretKey = SaslUtils.getSecretKey(this.client.getStormConf());
+ String secretKey;
+ name = client.name();
+ secretKey = client.secretKey();
+
if (secretKey != null) {
token = secretKey.getBytes();
}
- LOG.debug("SASL credentials for storm topology " + topologyName
+ LOG.debug("SASL credentials for storm topology " + name
+ " is " + secretKey);
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/jvm/backtype/storm/messaging/netty/Server.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/Server.java b/storm-core/src/jvm/backtype/storm/messaging/netty/Server.java
index e984144..7011b42 100644
--- a/storm-core/src/jvm/backtype/storm/messaging/netty/Server.java
+++ b/storm-core/src/jvm/backtype/storm/messaging/netty/Server.java
@@ -24,11 +24,13 @@ import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Collection;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
+import java.io.IOException;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.Channel;
@@ -36,17 +38,19 @@ import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.group.ChannelGroup;
import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import backtype.storm.Config;
+import backtype.storm.grouping.Load;
import backtype.storm.messaging.ConnectionWithStatus;
-import backtype.storm.messaging.IConnection;
import backtype.storm.messaging.TaskMessage;
import backtype.storm.metric.api.IStatefulObject;
+import backtype.storm.serialization.KryoValuesSerializer;
import backtype.storm.utils.Utils;
-class Server extends ConnectionWithStatus implements IStatefulObject {
+class Server extends ConnectionWithStatus implements IStatefulObject, ISaslServer {
private static final Logger LOG = LoggerFactory.getLogger(Server.class);
@SuppressWarnings("rawtypes")
@@ -56,7 +60,6 @@ class Server extends ConnectionWithStatus implements IStatefulObject {
private final AtomicInteger messagesDequeued = new AtomicInteger(0);
private final AtomicInteger[] pendingMessages;
-
// Create multiple queues for incoming messages. The size equals the number of receiver threads.
// For message which is sent to same task, it will be stored in the same queue to preserve the message order.
private LinkedBlockingQueue<ArrayList<TaskMessage>>[] message_queue;
@@ -71,12 +74,13 @@ class Server extends ConnectionWithStatus implements IStatefulObject {
private volatile boolean closing = false;
List<TaskMessage> closeMessage = Arrays.asList(new TaskMessage(-1, null));
-
+ private KryoValuesSerializer _ser;
@SuppressWarnings("rawtypes")
Server(Map storm_conf, int port) {
this.storm_conf = storm_conf;
this.port = port;
+ _ser = new KryoValuesSerializer(storm_conf);
queueCount = Utils.getInt(storm_conf.get(Config.WORKER_RECEIVER_THREAD_COUNT), 1);
roundRobinQueueId = 0;
@@ -94,8 +98,8 @@ class Server extends ConnectionWithStatus implements IStatefulObject {
int backlog = Utils.getInt(storm_conf.get(Config.STORM_MESSAGING_NETTY_SOCKET_BACKLOG), 500);
int maxWorkers = Utils.getInt(storm_conf.get(Config.STORM_MESSAGING_NETTY_SERVER_WORKER_THREADS));
- ThreadFactory bossFactory = new NettyRenameThreadFactory(name() + "-boss");
- ThreadFactory workerFactory = new NettyRenameThreadFactory(name() + "-worker");
+ ThreadFactory bossFactory = new NettyRenameThreadFactory(netty_name() + "-boss");
+ ThreadFactory workerFactory = new NettyRenameThreadFactory(netty_name() + "-worker");
if (maxWorkers > 0) {
factory = new NioServerSocketChannelFactory(Executors.newCachedThreadPool(bossFactory),
@@ -105,7 +109,7 @@ class Server extends ConnectionWithStatus implements IStatefulObject {
Executors.newCachedThreadPool(workerFactory));
}
- LOG.info("Create Netty Server " + name() + ", buffer_size: " + buffer_size + ", maxWorkers: " + maxWorkers);
+ LOG.info("Create Netty Server " + netty_name() + ", buffer_size: " + buffer_size + ", maxWorkers: " + maxWorkers);
bootstrap = new ServerBootstrap(factory);
bootstrap.setOption("child.tcpNoDelay", true);
@@ -181,7 +185,6 @@ class Server extends ConnectionWithStatus implements IStatefulObject {
}
}
-
/**
* enqueue a received message
* @throws InterruptedException
@@ -216,14 +219,12 @@ class Server extends ConnectionWithStatus implements IStatefulObject {
if ((flags & 0x01) == 0x01) {
//non-blocking
ret = message_queue[queueId].poll();
- }
- else {
+ } else {
try {
ArrayList<TaskMessage> request = message_queue[queueId].take();
LOG.debug("request to be processed: {}", request);
ret = request;
- }
- catch (InterruptedException e) {
+ } catch (InterruptedException e) {
LOG.info("exception within msg receiving", e);
ret = null;
}
@@ -249,7 +250,7 @@ class Server extends ConnectionWithStatus implements IStatefulObject {
* close a channel
* @param channel
*/
- protected void closeChannel(Channel channel) {
+ public void closeChannel(Channel channel) {
channel.close().awaitUninterruptibly();
allChannels.remove(channel);
}
@@ -265,15 +266,33 @@ class Server extends ConnectionWithStatus implements IStatefulObject {
}
}
+ @Override
+ public void sendLoadMetrics(Map<Integer, Double> taskToLoad) {
+ try {
+ MessageBatch mb = new MessageBatch(1);
+ mb.add(new TaskMessage(-1, _ser.serialize(Arrays.asList((Object)taskToLoad))));
+ allChannels.write(mb);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Map<Integer, Load> getLoad(Collection<Integer> tasks) {
+ throw new RuntimeException("Server connection cannot get load");
+ }
+
+ @Override
public void send(int task, byte[] message) {
throw new UnsupportedOperationException("Server connection should not send any messages");
}
+ @Override
public void send(Iterator<TaskMessage> msgs) {
throw new UnsupportedOperationException("Server connection should not send any messages");
}
- public String name() {
+ public String netty_name() {
return "Netty-server-localhost-" + port;
}
@@ -306,7 +325,7 @@ class Server extends ConnectionWithStatus implements IStatefulObject {
}
public Object getState() {
- LOG.info("Getting metrics for server on port {}", port);
+ LOG.debug("Getting metrics for server on port {}", port);
HashMap<String, Object> ret = new HashMap<String, Object>();
ret.put("dequeuedMessages", messagesDequeued.getAndSet(0));
ArrayList<Integer> pending = new ArrayList<Integer>(pendingMessages.length);
@@ -330,8 +349,30 @@ class Server extends ConnectionWithStatus implements IStatefulObject {
return ret;
}
- @Override public String toString() {
- return String.format("Netty server listening on port %s", port);
+ /** Implementing IServer. **/
+ public void channelConnected(Channel c) {
+ addChannel(c);
+ }
+
+ public void received(Object message, String remote, Channel channel) throws InterruptedException {
+ List<TaskMessage>msgs = (List<TaskMessage>)message;
+ enqueue(msgs, remote);
+ }
+
+ public String name() {
+ return (String)storm_conf.get(Config.TOPOLOGY_NAME);
+ }
+
+ public String secretKey() {
+ return SaslUtils.getSecretKey(storm_conf);
}
+ public void authenticated(Channel c) {
+ return;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("Netty server listening on port %s", port);
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/jvm/backtype/storm/messaging/netty/StormClientHandler.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/StormClientHandler.java b/storm-core/src/jvm/backtype/storm/messaging/netty/StormClientHandler.java
index 2d25001..877b6d8 100644
--- a/storm-core/src/jvm/backtype/storm/messaging/netty/StormClientHandler.java
+++ b/storm-core/src/jvm/backtype/storm/messaging/netty/StormClientHandler.java
@@ -17,7 +17,20 @@
*/
package backtype.storm.messaging.netty;
+import backtype.storm.messaging.TaskMessage;
+import backtype.storm.serialization.KryoValuesDeserializer;
+
import java.net.ConnectException;
+import java.util.Map;
+import java.util.List;
+import java.io.IOException;
+
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.channel.ChannelStateEvent;
+import org.jboss.netty.channel.ExceptionEvent;
+import org.jboss.netty.channel.MessageEvent;
+import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.channel.*;
import org.slf4j.Logger;
@@ -26,9 +39,41 @@ import org.slf4j.LoggerFactory;
public class StormClientHandler extends SimpleChannelUpstreamHandler {
private static final Logger LOG = LoggerFactory.getLogger(StormClientHandler.class);
private Client client;
-
- StormClientHandler(Client client) {
+ private KryoValuesDeserializer _des;
+
+ StormClientHandler(Client client, Map conf) {
this.client = client;
+ _des = new KryoValuesDeserializer(conf);
+ }
+
+ @Override
+ public void messageReceived(ChannelHandlerContext ctx, MessageEvent event) {
+ //examine the response message from server
+ Object message = event.getMessage();
+ if (message instanceof ControlMessage) {
+ ControlMessage msg = (ControlMessage)message;
+ if (msg==ControlMessage.FAILURE_RESPONSE) {
+ LOG.info("failure response:{}", msg);
+ }
+ } else if (message instanceof List) {
+ try {
+ //This should be the metrics, and there should only be one of them
+ List<TaskMessage> list = (List<TaskMessage>)message;
+ if (list.size() < 1) throw new RuntimeException("Didn't see enough load metrics ("+client.getDstAddress()+") "+list);
+ if (list.size() != 1) LOG.warn("Messages are not being delivered fast enough, got "+list.size()+" metrics messages at once("+client.getDstAddress()+")");
+ TaskMessage tm = ((List<TaskMessage>)message).get(list.size() - 1);
+ if (tm.task() != -1) throw new RuntimeException("Metrics messages are sent to the system task ("+client.getDstAddress()+") "+tm);
+ List metrics = _des.deserialize(tm.message());
+ if (metrics.size() < 1) throw new RuntimeException("No metrics data in the metrics message ("+client.getDstAddress()+") "+metrics);
+ if (!(metrics.get(0) instanceof Map)) throw new RuntimeException("The metrics did not have a map in the first slot ("+client.getDstAddress()+") "+metrics);
+ client.setLoadMetrics((Map<Integer, Double>)metrics.get(0));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ throw new RuntimeException("Don't know how to handle a message of type "
+ + message + " (" + client.getDstAddress() + ")");
+ }
}
@Override
@@ -40,7 +85,7 @@ public class StormClientHandler extends SimpleChannelUpstreamHandler {
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent event) {
Throwable cause = event.getCause();
if (!(cause instanceof ConnectException)) {
- LOG.info("Connection failed " + client.dstAddressPrefixedName, cause);
+ LOG.info("Connection to "+client.getDstAddress()+" failed:", cause);
}
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/src/jvm/backtype/storm/messaging/netty/StormClientPipelineFactory.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/StormClientPipelineFactory.java b/storm-core/src/jvm/backtype/storm/messaging/netty/StormClientPipelineFactory.java
index 4be06cd..6158eef 100644
--- a/storm-core/src/jvm/backtype/storm/messaging/netty/StormClientPipelineFactory.java
+++ b/storm-core/src/jvm/backtype/storm/messaging/netty/StormClientPipelineFactory.java
@@ -22,12 +22,15 @@ import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import backtype.storm.Config;
+import java.util.Map;
class StormClientPipelineFactory implements ChannelPipelineFactory {
private Client client;
+ private Map conf;
- StormClientPipelineFactory(Client client) {
+ StormClientPipelineFactory(Client client, Map conf) {
this.client = client;
+ this.conf = conf;
}
public ChannelPipeline getPipeline() throws Exception {
@@ -39,15 +42,15 @@ class StormClientPipelineFactory implements ChannelPipelineFactory {
// Encoder
pipeline.addLast("encoder", new MessageEncoder());
- boolean isNettyAuth = (Boolean) this.client.getStormConf().get(Config.STORM_MESSAGING_NETTY_AUTHENTICATION);
+ boolean isNettyAuth = (Boolean) conf
+ .get(Config.STORM_MESSAGING_NETTY_AUTHENTICATION);
if (isNettyAuth) {
// Authenticate: Removed after authentication completes
pipeline.addLast("saslClientHandler", new SaslStormClientHandler(
client));
}
// business logic.
- pipeline.addLast("handler", new StormClientHandler(client));
-
+ pipeline.addLast("handler", new StormClientHandler(client, conf));
return pipeline;
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/test/clj/backtype/storm/grouping_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/backtype/storm/grouping_test.clj b/storm-core/test/clj/backtype/storm/grouping_test.clj
index 2bfa066..f914591 100644
--- a/storm-core/test/clj/backtype/storm/grouping_test.clj
+++ b/storm-core/test/clj/backtype/storm/grouping_test.clj
@@ -17,27 +17,85 @@
(:use [clojure test])
(:import [backtype.storm.testing TestWordCounter TestWordSpout TestGlobalCount TestAggregatesCounter TestWordBytesCounter NGrouping]
[backtype.storm.generated JavaObject JavaObjectArg])
- (:use [backtype.storm testing clojure])
- (:use [backtype.storm.daemon common])
+ (:import [backtype.storm.grouping LoadMapping])
+ (:use [backtype.storm testing clojure log config])
+ (:use [backtype.storm.daemon common executor])
(:require [backtype.storm [thrift :as thrift]]))
(deftest test-shuffle
+ (let [shuffle-fn (mk-shuffle-grouper [(int 1) (int 2)] {TOPOLOGY-DISABLE-LOADAWARE-MESSAGING true} nil "comp" "stream")
+ num-messages 100000
+ min-prcnt (int (* num-messages 0.49))
+ max-prcnt (int (* num-messages 0.51))
+ data [1 2]
+ freq (frequencies (for [x (range 0 num-messages)] (shuffle-fn (int 1) data nil)))
+ load1 (.get freq [(int 1)])
+ load2 (.get freq [(int 2)])]
+ (log-message "FREQ:" freq)
+ (is (>= load1 min-prcnt))
+ (is (<= load1 max-prcnt))
+ (is (>= load2 min-prcnt))
+ (is (<= load2 max-prcnt))))
+
+(deftest test-shuffle-load-even
+ (let [shuffle-fn (mk-shuffle-grouper [(int 1) (int 2)] {} nil "comp" "stream")
+ num-messages 100000
+ min-prcnt (int (* num-messages 0.49))
+ max-prcnt (int (* num-messages 0.51))
+ load (LoadMapping.)
+ _ (.setLocal load {(int 1) 0.0 (int 2) 0.0})
+ data [1 2]
+ freq (frequencies (for [x (range 0 num-messages)] (shuffle-fn (int 1) data load)))
+ load1 (.get freq [(int 1)])
+ load2 (.get freq [(int 2)])]
+ (log-message "FREQ:" freq)
+ (is (>= load1 min-prcnt))
+ (is (<= load1 max-prcnt))
+ (is (>= load2 min-prcnt))
+ (is (<= load2 max-prcnt))))
+
+(deftest test-shuffle-load-uneven
+ (let [shuffle-fn (mk-shuffle-grouper [(int 1) (int 2)] {} nil "comp" "stream")
+ num-messages 100000
+ min1-prcnt (int (* num-messages 0.32))
+ max1-prcnt (int (* num-messages 0.34))
+ min2-prcnt (int (* num-messages 0.65))
+ max2-prcnt (int (* num-messages 0.67))
+ load (LoadMapping.)
+ _ (.setLocal load {(int 1) 0.5 (int 2) 0.0})
+ data [1 2]
+ freq (frequencies (for [x (range 0 num-messages)] (shuffle-fn (int 1) data load)))
+ load1 (.get freq [(int 1)])
+ load2 (.get freq [(int 2)])]
+ (log-message "FREQ:" freq)
+ (is (>= load1 min1-prcnt))
+ (is (<= load1 max1-prcnt))
+ (is (>= load2 min2-prcnt))
+ (is (<= load2 max2-prcnt))))
+
+(deftest test-field
(with-simulated-time-local-cluster [cluster :supervisors 4]
- (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 [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)
})
- results (complete-topology cluster
- topology
- ;; important for test that
- ;; #tuples = multiple of 4 and 6
- :mock-sources {"1" (->> [["a"] ["b"]]
- (repeat 12)
- (apply concat))})]
- (is (ms= (apply concat (repeat 6 [[1] [2] [3] [4]]))
- (read-tuples results "2")))
- )))
+ results (complete-topology
+ cluster
+ topology
+ :mock-sources {"1" (->> [[(.getBytes "a")]
+ [(.getBytes "b")]]
+ (repeat (* spout-phint bolt-phint))
+ (apply concat))})]
+ (is (ms= (apply concat
+ (for [value '("a" "b")
+ sum (range 1 (inc (* spout-phint bolt-phint)))]
+ [[value sum]]))
+ (read-tuples results "2"))))))
(deftest test-field
(with-simulated-time-local-cluster [cluster :supervisors 4]
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/test/clj/backtype/storm/integration_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/backtype/storm/integration_test.clj b/storm-core/test/clj/backtype/storm/integration_test.clj
index ca17d6e..cc0208d 100644
--- a/storm-core/test/clj/backtype/storm/integration_test.clj
+++ b/storm-core/test/clj/backtype/storm/integration_test.clj
@@ -62,13 +62,13 @@
(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" :shuffle} emit-task-id
+ {"2" (thrift/mk-bolt-spec {"1" :all} emit-task-id
:parallelism-hint 3
:conf {TOPOLOGY-TASKS 6})
})
results (complete-topology cluster
topology
- :mock-sources {"1" [["a"] ["a"] ["a"] ["a"] ["a"] ["a"]]})]
+ :mock-sources {"1" [["a"]]})]
(is (ms= [[0] [1] [2] [3] [4] [5]]
(read-tuples results "2")))
)))
http://git-wip-us.apache.org/repos/asf/storm/blob/af64a3a2/storm-core/test/clj/backtype/storm/messaging/netty_integration_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/backtype/storm/messaging/netty_integration_test.clj b/storm-core/test/clj/backtype/storm/messaging/netty_integration_test.clj
index a918a4b..8081ca5 100644
--- a/storm-core/test/clj/backtype/storm/messaging/netty_integration_test.clj
+++ b/storm-core/test/clj/backtype/storm/messaging/netty_integration_test.clj
@@ -54,5 +54,4 @@
["a"] ["b"]
]}
)]
- (is (ms= (apply concat (repeat 6 [[1] [2] [3] [4]]))
- (read-tuples results "2"))))))
+ (is (= (* 6 4) (.size (read-tuples results "2")))))))