You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2016/11/23 19:39:49 UTC
[1/6] storm git commit: remove array shuffle from ShuffleGrouping
when counter resets to zero
Repository: storm
Updated Branches:
refs/heads/1.0.x-branch 916250f5f -> 5938f2cba
remove array shuffle from ShuffleGrouping when counter resets to zero
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/980af808
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/980af808
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/980af808
Branch: refs/heads/1.0.x-branch
Commit: 980af8083aaf185fad3db476dd765abe4ea9cda4
Parents: 55f5567
Author: Kevin Peek <kp...@salesforce.com>
Authored: Mon Nov 21 13:53:33 2016 -0500
Committer: Kevin Peek <kp...@salesforce.com>
Committed: Mon Nov 21 13:53:33 2016 -0500
----------------------------------------------------------------------
.../apache/storm/grouping/ShuffleGrouping.java | 14 +--
.../storm/grouping/ShuffleGroupingTest.java | 124 +++++++++++++++++++
2 files changed, 128 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/980af808/storm-core/src/jvm/org/apache/storm/grouping/ShuffleGrouping.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/grouping/ShuffleGrouping.java b/storm-core/src/jvm/org/apache/storm/grouping/ShuffleGrouping.java
index e8b1d71..10bfb2a 100644
--- a/storm-core/src/jvm/org/apache/storm/grouping/ShuffleGrouping.java
+++ b/storm-core/src/jvm/org/apache/storm/grouping/ShuffleGrouping.java
@@ -17,17 +17,13 @@
*/
package org.apache.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 org.apache.storm.generated.GlobalStreamId;
import org.apache.storm.task.WorkerTopologyContext;
+import java.io.Serializable;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
public class ShuffleGrouping implements CustomStreamGrouping, Serializable {
private Random random;
private ArrayList<List<Integer>> choices;
@@ -54,8 +50,6 @@ public class ShuffleGrouping implements CustomStreamGrouping, Serializable {
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
http://git-wip-us.apache.org/repos/asf/storm/blob/980af808/storm-core/test/jvm/org/apache/storm/grouping/ShuffleGroupingTest.java
----------------------------------------------------------------------
diff --git a/storm-core/test/jvm/org/apache/storm/grouping/ShuffleGroupingTest.java b/storm-core/test/jvm/org/apache/storm/grouping/ShuffleGroupingTest.java
new file mode 100644
index 0000000..2a0b87f
--- /dev/null
+++ b/storm-core/test/jvm/org/apache/storm/grouping/ShuffleGroupingTest.java
@@ -0,0 +1,124 @@
+package org.apache.storm.grouping;
+
+import com.google.common.collect.Lists;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.junit.Test;
+
+import java.util.List;
+import java.util.concurrent.*;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.mock;
+
+public class ShuffleGroupingTest {
+
+ /**
+ * Tests that we round robbin correctly using ShuffleGrouping implementation.
+ * */
+ @Test
+ public void testShuffleGrouping() {
+ final int numTasks = 6;
+
+ final ShuffleGrouping grouper = new ShuffleGrouping();
+
+ // Task Id not used, so just pick a static value
+ final int inputTaskId = 100;
+
+ // Define our taskIds
+ final List<Integer> availableTaskIds = Lists.newArrayList();
+ for (int i = 0; i < numTasks; i++) {
+ availableTaskIds.add(i);
+ }
+ WorkerTopologyContext context = mock(WorkerTopologyContext.class);
+ grouper.prepare(context, null, availableTaskIds);
+
+ // Keep track of how many times we see each taskId
+ int[] taskCounts = new int[numTasks];
+ for (int i = 1; i <= 30000; i++) {
+ List<Integer> taskIds = grouper.chooseTasks(inputTaskId, Lists.newArrayList());
+
+ // Validate a single task id return
+ assertNotNull("Not null taskId list returned", taskIds);
+ assertEquals("Single task Id returned", 1, taskIds.size());
+
+ int taskId = taskIds.get(0);
+
+ assertTrue("TaskId should exist", taskId >= 0 && taskId < numTasks);
+ taskCounts[taskId]++;
+ }
+
+ for (int i = 0; i < numTasks; i++) {
+ assertEquals("Distribution should be even for all nodes", 5000, taskCounts[i]);
+ }
+ }
+
+ /**
+ * Tests that we round robbin correctly with multiple threads using ShuffleGrouping implementation.
+ */
+ @Test
+ public void testShuffleGroupMultiThreaded() throws InterruptedException, ExecutionException {
+ final int numTasks = 6;
+ final int groupingExecutionsPerThread = 30000;
+ final int numThreads = 10;
+
+ final CustomStreamGrouping grouper = new ShuffleGrouping();
+
+ // Task Id not used, so just pick a static value
+ final int inputTaskId = 100;
+ // Define our taskIds - the test expects these to be incrementing by one up from zero
+ final List<Integer> availableTaskIds = Lists.newArrayList();
+ for (int i = 0; i < numTasks; i++) {
+ availableTaskIds.add(i);
+ }
+
+ final WorkerTopologyContext context = mock(WorkerTopologyContext.class);
+
+ // Call prepare with our available taskIds
+ grouper.prepare(context, null, availableTaskIds);
+
+ List<Callable<int[]>> threadTasks = Lists.newArrayList();
+ for (int x=0; x < numThreads; x++) {
+ Callable<int[]> threadTask = new Callable<int[]>() {
+ @Override
+ public int[] call() throws Exception {
+ int[] taskCounts = new int[availableTaskIds.size()];
+ for (int i = 1; i <= groupingExecutionsPerThread; i++) {
+ List<Integer> taskIds = grouper.chooseTasks(inputTaskId, Lists.newArrayList());
+
+ // Validate a single task id return
+ assertNotNull("Not null taskId list returned", taskIds);
+ assertEquals("Single task Id returned", 1, taskIds.size());
+
+ int taskId = taskIds.get(0);
+
+ assertTrue("TaskId should exist", taskId >= 0 && taskId < availableTaskIds.size());
+ taskCounts[taskId]++;
+ }
+ return taskCounts;
+ }
+ };
+
+ // Add to our collection.
+ threadTasks.add(threadTask);
+ }
+
+ ExecutorService executor = Executors.newFixedThreadPool(threadTasks.size());
+ List<Future<int[]>> taskResults = executor.invokeAll(threadTasks);
+
+ // Wait for all tasks to complete
+ int[] taskIdTotals = new int[numTasks];
+ for (Future taskResult: taskResults) {
+ while (!taskResult.isDone()) {
+ Thread.sleep(1000);
+ }
+ int[] taskDistributions = (int[]) taskResult.get();
+ for (int i = 0; i < taskDistributions.length; i++) {
+ taskIdTotals[i] += taskDistributions[i];
+ }
+ }
+
+ for (int i = 0; i < numTasks; i++) {
+ assertEquals(numThreads * groupingExecutionsPerThread / numTasks, taskIdTotals[i]);
+ }
+ }
+}
\ No newline at end of file
[4/6] storm git commit: clean up wildcard imports
Posted by bo...@apache.org.
clean up wildcard imports
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/8132d13c
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/8132d13c
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/8132d13c
Branch: refs/heads/1.0.x-branch
Commit: 8132d13c0be2aa8045b5b8dfe122724a7f7c3559
Parents: 991d82a
Author: Kevin Peek <kp...@salesforce.com>
Authored: Mon Nov 21 15:06:45 2016 -0500
Committer: Kevin Peek <kp...@salesforce.com>
Committed: Mon Nov 21 15:06:45 2016 -0500
----------------------------------------------------------------------
.../src/jvm/org/apache/storm/grouping/ShuffleGrouping.java | 6 +++++-
1 file changed, 5 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/8132d13c/storm-core/src/jvm/org/apache/storm/grouping/ShuffleGrouping.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/grouping/ShuffleGrouping.java b/storm-core/src/jvm/org/apache/storm/grouping/ShuffleGrouping.java
index 10bfb2a..ad8014c 100644
--- a/storm-core/src/jvm/org/apache/storm/grouping/ShuffleGrouping.java
+++ b/storm-core/src/jvm/org/apache/storm/grouping/ShuffleGrouping.java
@@ -21,7 +21,11 @@ import org.apache.storm.generated.GlobalStreamId;
import org.apache.storm.task.WorkerTopologyContext;
import java.io.Serializable;
-import java.util.*;
+import java.util.List;
+import java.util.Random;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.concurrent.atomic.AtomicInteger;
public class ShuffleGrouping implements CustomStreamGrouping, Serializable {
[3/6] storm git commit: clean up wildcard imports
Posted by bo...@apache.org.
clean up wildcard imports
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/991d82a4
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/991d82a4
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/991d82a4
Branch: refs/heads/1.0.x-branch
Commit: 991d82a4d18cdc34c5628c25ee64e5ff511134f2
Parents: 704b840
Author: Kevin Peek <kp...@salesforce.com>
Authored: Mon Nov 21 15:05:29 2016 -0500
Committer: Kevin Peek <kp...@salesforce.com>
Committed: Mon Nov 21 15:05:29 2016 -0500
----------------------------------------------------------------------
.../org/apache/storm/grouping/ShuffleGroupingTest.java | 12 +++++++++---
1 file changed, 9 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/991d82a4/storm-core/test/jvm/org/apache/storm/grouping/ShuffleGroupingTest.java
----------------------------------------------------------------------
diff --git a/storm-core/test/jvm/org/apache/storm/grouping/ShuffleGroupingTest.java b/storm-core/test/jvm/org/apache/storm/grouping/ShuffleGroupingTest.java
index 6756078..1f628b9 100644
--- a/storm-core/test/jvm/org/apache/storm/grouping/ShuffleGroupingTest.java
+++ b/storm-core/test/jvm/org/apache/storm/grouping/ShuffleGroupingTest.java
@@ -22,9 +22,15 @@ import org.apache.storm.task.WorkerTopologyContext;
import org.junit.Test;
import java.util.List;
-import java.util.concurrent.*;
-
-import static org.junit.Assert.*;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.Executors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
public class ShuffleGroupingTest {
[2/6] storm git commit: add license info to new test
Posted by bo...@apache.org.
add license info to new test
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/704b8403
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/704b8403
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/704b8403
Branch: refs/heads/1.0.x-branch
Commit: 704b8403b46816b47c90be75d4617247fe2b67c6
Parents: 980af80
Author: Kevin Peek <kp...@salesforce.com>
Authored: Mon Nov 21 14:08:31 2016 -0500
Committer: Kevin Peek <kp...@salesforce.com>
Committed: Mon Nov 21 14:08:31 2016 -0500
----------------------------------------------------------------------
.../apache/storm/grouping/ShuffleGroupingTest.java | 17 +++++++++++++++++
1 file changed, 17 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/704b8403/storm-core/test/jvm/org/apache/storm/grouping/ShuffleGroupingTest.java
----------------------------------------------------------------------
diff --git a/storm-core/test/jvm/org/apache/storm/grouping/ShuffleGroupingTest.java b/storm-core/test/jvm/org/apache/storm/grouping/ShuffleGroupingTest.java
index 2a0b87f..6756078 100644
--- a/storm-core/test/jvm/org/apache/storm/grouping/ShuffleGroupingTest.java
+++ b/storm-core/test/jvm/org/apache/storm/grouping/ShuffleGroupingTest.java
@@ -1,3 +1,20 @@
+/**
+ * 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 org.apache.storm.grouping;
import com.google.common.collect.Lists;
[6/6] storm git commit: STORM-2210: Changelog
Posted by bo...@apache.org.
STORM-2210: Changelog
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/5938f2cb
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/5938f2cb
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/5938f2cb
Branch: refs/heads/1.0.x-branch
Commit: 5938f2cbaf0a86479f2f65d8d4d5ec2f99827fa9
Parents: 9ef74c1
Author: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Authored: Wed Nov 23 13:36:24 2016 -0600
Committer: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Committed: Wed Nov 23 13:36:24 2016 -0600
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/5938f2cb/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index cf5854e..77164e7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,5 @@
## 1.0.3
+ * STORM-2210: remove array shuffle from ShuffleGrouping
* STORM-2052: Kafka Spout - New Client API - Performance Improvements
* storm-2205: Racecondition in getting nimbus summaries while ZK connections are reconnected
* STORM-2198: perform RotationAction when stopping HdfsBolt
[5/6] storm git commit: Merge branch 'STORM-2210' of
https://github.com/kevpeek/storm into STORM-2210-1.0.x
Posted by bo...@apache.org.
Merge branch 'STORM-2210' of https://github.com/kevpeek/storm into STORM-2210-1.0.x
STORM-2210: remove array shuffle from ShuffleGrouping
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/9ef74c13
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/9ef74c13
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/9ef74c13
Branch: refs/heads/1.0.x-branch
Commit: 9ef74c132bf62e4735839abd83b5aadc818a3cfa
Parents: 916250f 8132d13
Author: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Authored: Wed Nov 23 13:35:57 2016 -0600
Committer: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Committed: Wed Nov 23 13:35:57 2016 -0600
----------------------------------------------------------------------
.../apache/storm/grouping/ShuffleGrouping.java | 12 +-
.../storm/grouping/ShuffleGroupingTest.java | 147 +++++++++++++++++++
2 files changed, 152 insertions(+), 7 deletions(-)
----------------------------------------------------------------------