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 2015/11/17 16:46:19 UTC
[2/6] storm git commit: replace HashSet with LinkedHashSet in
TridentTopology.java
replace HashSet with LinkedHashSet in TridentTopology.java
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/3dbfa10c
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/3dbfa10c
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/3dbfa10c
Branch: refs/heads/master
Commit: 3dbfa10c26a71d33f35037edc4e57a47f50a11f8
Parents: f8118e0
Author: wangjiasheng <wa...@xiaomi.com>
Authored: Fri Nov 13 16:02:32 2015 +0800
Committer: wangjiasheng <wa...@xiaomi.com>
Committed: Fri Nov 13 16:02:32 2015 +0800
----------------------------------------------------------------------
.../src/jvm/storm/trident/TridentTopology.java | 45 +++++---------------
.../jvm/storm/trident/graph/GraphGrouper.java | 9 ++--
2 files changed, 14 insertions(+), 40 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/3dbfa10c/storm-core/src/jvm/storm/trident/TridentTopology.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/storm/trident/TridentTopology.java b/storm-core/src/jvm/storm/trident/TridentTopology.java
index 519a020..aec6eb8 100644
--- a/storm-core/src/jvm/storm/trident/TridentTopology.java
+++ b/storm-core/src/jvm/storm/trident/TridentTopology.java
@@ -30,17 +30,8 @@ import backtype.storm.tuple.Fields;
import backtype.storm.utils.Utils;
import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeMap;
+import java.util.*;
+
import org.jgrapht.DirectedGraph;
import org.jgrapht.UndirectedGraph;
import org.jgrapht.alg.ConnectivityInspector;
@@ -93,7 +84,7 @@ public class TridentTopology {
// is it too expensive to do a batch per drpc request?
DefaultDirectedGraph<Node, IndexedEdge> _graph;
- Map<String, List<Node>> _colocate = new HashMap();
+ Map<String, List<Node>> _colocate = new LinkedHashMap();
UniqueIdGen _gen;
public TridentTopology() {
@@ -173,7 +164,7 @@ public class TridentTopology {
}
public Stream multiReduce(Stream s1, Stream s2, MultiReducer function, Fields outputFields) {
- return multiReduce(Arrays.asList(s1, s2), function, outputFields);
+ return multiReduce(Arrays.asList(s1, s2), function, outputFields);
}
public Stream multiReduce(Fields inputFields1, Stream s1, Fields inputFields2, Stream s2, MultiReducer function, Fields outputFields) {
@@ -275,7 +266,7 @@ public class TridentTopology {
List<SpoutNode> spoutNodes = new ArrayList<SpoutNode>();
// can be regular nodes (static state) or processor nodes
- Set<Node> boltNodes = new HashSet<Node>();
+ Set<Node> boltNodes = new LinkedHashSet<Node>();
for(Node n: graph.vertexSet()) {
if(n instanceof SpoutNode) {
spoutNodes.add((SpoutNode) n);
@@ -285,7 +276,7 @@ public class TridentTopology {
}
- Set<Group> initialGroups = new HashSet<Group>();
+ Set<Group> initialGroups = new LinkedHashSet<Group>();
for(List<Node> colocate: _colocate.values()) {
Group g = new Group(graph, colocate);
boltNodes.removeAll(colocate);
@@ -528,29 +519,15 @@ public class TridentTopology {
private static Map<Group, String> genBoltIds(Collection<Group> groups) {
Map<Group, String> ret = new HashMap();
- //sort by group name
- final Map<Group, String> groupNames = new HashMap();
- for (Group g : groups) {
- String name = getGroupName(g);
- groupNames.put(g, name == null ? "" : name);
- }
- List<Group> groupList = new ArrayList(groups);
- Collections.sort(groupList, new Comparator<Group>() {
- @Override
- public int compare(Group o1, Group o2) {
- return groupNames.get(o1).compareTo(groupNames.get(o2));
- }
- });
-
int ctr = 0;
- for (Group g : groupList) {
- if (!isSpoutGroup(g)) {
+ for(Group g: groups) {
+ if(!isSpoutGroup(g)) {
List<String> name = new ArrayList();
name.add("b");
name.add("" + ctr);
- String groupName = groupNames.get(g);
- if (groupName != null && !groupName.isEmpty()) {
- name.add(groupName);
+ String groupName = getGroupName(g);
+ if(groupName!=null && !groupName.isEmpty()) {
+ name.add(getGroupName(g));
}
ret.put(g, Utils.join(name, "-"));
ctr++;
http://git-wip-us.apache.org/repos/asf/storm/blob/3dbfa10c/storm-core/src/jvm/storm/trident/graph/GraphGrouper.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/storm/trident/graph/GraphGrouper.java b/storm-core/src/jvm/storm/trident/graph/GraphGrouper.java
index 94db077..5822ddc 100644
--- a/storm-core/src/jvm/storm/trident/graph/GraphGrouper.java
+++ b/storm-core/src/jvm/storm/trident/graph/GraphGrouper.java
@@ -17,11 +17,8 @@
*/
package storm.trident.graph;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
+
import org.jgrapht.DirectedGraph;
import storm.trident.planner.Node;
import storm.trident.util.IndexedEdge;
@@ -35,7 +32,7 @@ public class GraphGrouper {
public GraphGrouper(DirectedGraph<Node, IndexedEdge> graph, Collection<Group> initialGroups) {
this.graph = graph;
- this.currGroups = new HashSet(initialGroups);
+ this.currGroups = new LinkedHashSet(initialGroups);
reindex();
}