storm-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kabh...@apache.org
Subject [04/10] storm git commit: Correctly hash byte array tuple values
Date Tue, 04 Aug 2015 06:21:23 GMT
Correctly hash byte array tuple values


Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/0fe6accf
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/0fe6accf
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/0fe6accf

Branch: refs/heads/0.10.x-branch
Commit: 0fe6accfdfc8cbedfb5529fdba37193edcd69a26
Parents: ddb9e4c
Author: Derek Dagit <derekd@yahoo-inc.com>
Authored: Fri Jul 17 18:45:39 2015 -0500
Committer: Jungtaek Lim <kabhwan@gmail.com>
Committed: Tue Aug 4 13:07:38 2015 +0900

----------------------------------------------------------------------
 .../src/clj/backtype/storm/daemon/task.clj      |  1 -
 storm-core/src/clj/backtype/storm/tuple.clj     |  7 +-
 .../storm/grouping/PartialKeyGrouping.java      |  7 +-
 .../storm/testing/TestWordBytesCounter.java     | 27 ++++++++
 .../backtype/storm/testing/TestWordCounter.java |  6 +-
 .../backtype/storm/testing/WordBytesCount.java  | 35 ++++++++++
 .../test/clj/backtype/storm/grouping_test.clj   | 67 ++++++++++++--------
 7 files changed, 118 insertions(+), 32 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/storm/blob/0fe6accf/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 3bc1d05..9cf2b85 100644
--- a/storm-core/src/clj/backtype/storm/daemon/task.clj
+++ b/storm-core/src/clj/backtype/storm/daemon/task.clj
@@ -27,7 +27,6 @@
   (:import [backtype.storm.spout ShellSpout])
   (:import [java.util Collection List ArrayList])
   (:require [backtype.storm
-             [tuple :as tuple]
              [thrift :as thrift]
              [stats :as stats]])
   (:require [backtype.storm.daemon.builtin-metrics :as builtin-metrics]))

http://git-wip-us.apache.org/repos/asf/storm/blob/0fe6accf/storm-core/src/clj/backtype/storm/tuple.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/tuple.clj b/storm-core/src/clj/backtype/storm/tuple.clj
index a88b73b..790a823 100644
--- a/storm-core/src/clj/backtype/storm/tuple.clj
+++ b/storm-core/src/clj/backtype/storm/tuple.clj
@@ -15,8 +15,11 @@
 ;; limitations under the License.
 
 (ns backtype.storm.tuple
-  (:import [java.util List]))
+  (:import [java.util Arrays List]))
 
 (defn list-hash-code
   [^List alist]
-  (.hashCode alist))
+  (if (nil? alist)
+    1
+    (let [^"[Ljava.lang.Object;" array (.toArray alist)] ;; Object[]
+      (Arrays/deepHashCode array))))

http://git-wip-us.apache.org/repos/asf/storm/blob/0fe6accf/storm-core/src/jvm/backtype/storm/grouping/PartialKeyGrouping.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/grouping/PartialKeyGrouping.java b/storm-core/src/jvm/backtype/storm/grouping/PartialKeyGrouping.java
index d1f534b..456a8ef 100644
--- a/storm-core/src/jvm/backtype/storm/grouping/PartialKeyGrouping.java
+++ b/storm-core/src/jvm/backtype/storm/grouping/PartialKeyGrouping.java
@@ -20,6 +20,7 @@ package backtype.storm.grouping;
 import java.io.Serializable;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 import backtype.storm.generated.GlobalStreamId;
@@ -65,7 +66,11 @@ public class PartialKeyGrouping implements CustomStreamGrouping, Serializable
{
                 List<Object> selectedFields = outFields.select(fields, values);
                 ByteBuffer out = ByteBuffer.allocate(selectedFields.size() * 4);
                 for (Object o: selectedFields) {
-                    out.putInt(o.hashCode());
+                    if (o instanceof Object[]) {
+                        out.putInt(Arrays.deepHashCode((Object[])o));
+                    } else {
+                        out.putInt(o.hashCode());
+                    }
                 }
                 raw = out.array();
             } else {

http://git-wip-us.apache.org/repos/asf/storm/blob/0fe6accf/storm-core/src/jvm/backtype/storm/testing/TestWordBytesCounter.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/testing/TestWordBytesCounter.java b/storm-core/src/jvm/backtype/storm/testing/TestWordBytesCounter.java
new file mode 100644
index 0000000..e8a09a7
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/testing/TestWordBytesCounter.java
@@ -0,0 +1,27 @@
+/**
+ * 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.testing;
+
+import backtype.storm.tuple.Tuple;
+
+public class TestWordBytesCounter extends TestWordCounter {
+    @Override
+    protected String getTupleValue(Tuple t, int idx) {
+      return new String(t.getBinary(idx));
+    }
+}

http://git-wip-us.apache.org/repos/asf/storm/blob/0fe6accf/storm-core/src/jvm/backtype/storm/testing/TestWordCounter.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/testing/TestWordCounter.java b/storm-core/src/jvm/backtype/storm/testing/TestWordCounter.java
index 551b054..08b1397 100644
--- a/storm-core/src/jvm/backtype/storm/testing/TestWordCounter.java
+++ b/storm-core/src/jvm/backtype/storm/testing/TestWordCounter.java
@@ -38,9 +38,13 @@ public class TestWordCounter extends BaseBasicBolt {
     public void prepare(Map stormConf, TopologyContext context) {
         _counts = new HashMap<String, Integer>();
     }
+
+    protected String getTupleValue(Tuple t, int idx) {
+      return (String) t.getValues().get(idx);
+    }
     
     public void execute(Tuple input, BasicOutputCollector collector) {
-        String word = (String) input.getValues().get(0);
+        String word = getTupleValue(input, 0);
         int count = 0;
         if(_counts.containsKey(word)) {
             count = _counts.get(word);

http://git-wip-us.apache.org/repos/asf/storm/blob/0fe6accf/storm-core/src/jvm/backtype/storm/testing/WordBytesCount.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/testing/WordBytesCount.java b/storm-core/src/jvm/backtype/storm/testing/WordBytesCount.java
new file mode 100644
index 0000000..31fdca6
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/testing/WordBytesCount.java
@@ -0,0 +1,35 @@
+/**
+ * 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.testing;
+
+import backtype.storm.topology.BasicOutputCollector;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.topology.base.BaseBasicBolt;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class WordBytesCount extends WordCount {
+  @Override
+  protected String getWordFromTuple(Tuple tuple, int idx) {
+    return new String(tuple.getBinary(idx));
+  }
+}

http://git-wip-us.apache.org/repos/asf/storm/blob/0fe6accf/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 fc13b0c..9402842 100644
--- a/storm-core/test/clj/backtype/storm/grouping_test.clj
+++ b/storm-core/test/clj/backtype/storm/grouping_test.clj
@@ -15,40 +15,53 @@
 ;; limitations under the License.
 (ns backtype.storm.grouping-test
   (:use [clojure test])
-  (:import [backtype.storm.testing TestWordCounter TestWordSpout TestGlobalCount TestAggregatesCounter
NGrouping]
+  (: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])
   (:require [backtype.storm [thrift :as thrift]]))
 
-(deftest test-shuffle
+ (deftest test-shuffle
+   (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)
+                      })
+           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")))
+       )))
+
+(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"]
-                                                         ["a"] ["b"]
-                                                         ["a"] ["b"]
-                                                         ["a"] ["b"]
-                                                         ["a"] ["b"]
-                                                         ["a"] ["b"]
-                                                         ["a"] ["b"]
-                                                         ["a"] ["b"]
-                                                         ["a"] ["b"]
-                                                         ["a"] ["b"]
-                                                         ["a"] ["b"]
-                                                         ["a"] ["b"]
-                                                       ]}
-                                     )]
-      (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"))))))
 
 (defbolt id-bolt ["val"] [tuple collector]
   (emit-bolt! collector (.getValues tuple))


Mime
View raw message