This is an automated email from the ASF dual-hosted git repository. rzo1 pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/storm.git
commit 046cab5a8019b8de92523159886363e29c7d2cd9 Author: Richard Zowalla <[email protected]> AuthorDate: Thu Apr 16 20:07:25 2026 +0200 Upgrade to JGraphT 1.5.3 --- DEPENDENCY-LICENSES | 2 +- LICENSE-binary | 2 +- .../org/apache/storm/streams/ProcessorBolt.java | 4 ++-- .../storm/streams/ProcessorBoltDelegate.java | 10 +++++----- .../storm/streams/StatefulProcessorBolt.java | 4 ++-- .../org/apache/storm/streams/StreamBuilder.java | 14 ++++++------- .../jvm/org/apache/storm/streams/StreamUtil.java | 6 +++--- .../apache/storm/streams/StreamsEdgeFactory.java | 23 ---------------------- .../storm/streams/WindowedProcessorBolt.java | 4 ++-- .../org/apache/storm/trident/TridentTopology.java | 13 ++++++------ .../apache/storm/trident/graph/GraphGrouper.java | 6 +++--- .../jvm/org/apache/storm/trident/graph/Group.java | 8 ++++---- .../storm/trident/planner/SubtopologyBolt.java | 12 +++++------ .../storm/trident/util/ErrorEdgeFactory.java | 6 +++--- .../apache/storm/trident/util/TridentUtils.java | 8 ++++---- .../apache/storm/streams/ProcessorBoltTest.java | 6 +++--- .../storm/streams/StatefulProcessorBoltTest.java | 6 +++--- .../storm/streams/WindowedProcessorBoltTest.java | 6 +++--- 18 files changed, 57 insertions(+), 83 deletions(-) diff --git a/DEPENDENCY-LICENSES b/DEPENDENCY-LICENSES index 9a21ac8a8..8bb8f3a3e 100644 --- a/DEPENDENCY-LICENSES +++ b/DEPENDENCY-LICENSES @@ -405,7 +405,7 @@ List of third-party dependencies grouped by their license type. Eclipse Public License (EPL) 1.0, GNU Lesser General Public License Version 2.1, February 1999 - * JGraphT - Core (org.jgrapht:jgrapht-core:0.9.0 - http://www.jgrapht.org/jgrapht-core) + * JGraphT - Core (org.jgrapht:jgrapht-core:1.5.3 - http://www.jgrapht.org/jgrapht-core) Eclipse Public License 2.0, GNU General Public License, version 2 with the GNU Classpath Exception diff --git a/LICENSE-binary b/LICENSE-binary index da531828e..ecb7d6c5c 100644 --- a/LICENSE-binary +++ b/LICENSE-binary @@ -899,7 +899,7 @@ The license texts of these dependencies can be found in the licenses directory. Eclipse Public License (EPL) 1.0, GNU Lesser General Public License Version 2.1, February 1999 - * JGraphT - Core (org.jgrapht:jgrapht-core:0.9.0 - http://www.jgrapht.org/jgrapht-core) + * JGraphT - Core (org.jgrapht:jgrapht-core:1.5.3 - http://www.jgrapht.org/jgrapht-core) Eclipse Public License v. 2.0, GNU General Public License, version 2 with the GNU Classpath Exception diff --git a/storm-client/src/jvm/org/apache/storm/streams/ProcessorBolt.java b/storm-client/src/jvm/org/apache/storm/streams/ProcessorBolt.java index 559b40aca..5c8fc13cf 100644 --- a/storm-client/src/jvm/org/apache/storm/streams/ProcessorBolt.java +++ b/storm-client/src/jvm/org/apache/storm/streams/ProcessorBolt.java @@ -15,7 +15,7 @@ package org.apache.storm.streams; import java.util.List; import java.util.Map; import org.apache.storm.shade.com.google.common.collect.Multimap; -import org.apache.storm.shade.org.jgrapht.DirectedGraph; +import org.apache.storm.shade.org.jgrapht.Graph; import org.apache.storm.task.OutputCollector; import org.apache.storm.task.TopologyContext; import org.apache.storm.topology.OutputFieldsDeclarer; @@ -28,7 +28,7 @@ import org.apache.storm.tuple.Tuple; class ProcessorBolt extends BaseRichBolt implements StreamBolt { private final ProcessorBoltDelegate delegate; - ProcessorBolt(String id, DirectedGraph<Node, Edge> graph, List<ProcessorNode> nodes) { + ProcessorBolt(String id, Graph<Node, Edge> graph, List<ProcessorNode> nodes) { delegate = new ProcessorBoltDelegate(id, graph, nodes); } diff --git a/storm-client/src/jvm/org/apache/storm/streams/ProcessorBoltDelegate.java b/storm-client/src/jvm/org/apache/storm/streams/ProcessorBoltDelegate.java index f68746f12..ccb651100 100644 --- a/storm-client/src/jvm/org/apache/storm/streams/ProcessorBoltDelegate.java +++ b/storm-client/src/jvm/org/apache/storm/streams/ProcessorBoltDelegate.java @@ -26,8 +26,8 @@ import org.apache.storm.shade.com.google.common.collect.ArrayListMultimap; import org.apache.storm.shade.com.google.common.collect.HashBasedTable; import org.apache.storm.shade.com.google.common.collect.Multimap; import org.apache.storm.shade.com.google.common.collect.Table; -import org.apache.storm.shade.org.jgrapht.DirectedGraph; -import org.apache.storm.shade.org.jgrapht.graph.DirectedSubgraph; +import org.apache.storm.shade.org.jgrapht.Graph; +import org.apache.storm.shade.org.jgrapht.graph.AsSubgraph; import org.apache.storm.shade.org.jgrapht.traverse.TopologicalOrderIterator; import org.apache.storm.streams.processors.ChainedProcessorContext; import org.apache.storm.streams.processors.EmittingProcessorContext; @@ -45,7 +45,7 @@ import org.slf4j.LoggerFactory; class ProcessorBoltDelegate implements Serializable { private static final Logger LOG = LoggerFactory.getLogger(ProcessorBoltDelegate.class); private final String id; - private final DirectedGraph<Node, Edge> graph; + private final Graph<Node, Edge> graph; private final List<ProcessorNode> nodes; private final List<ProcessorNode> outgoingProcessors = new ArrayList<>(); private final Set<EmittingProcessorContext> emittingProcessorContexts = new HashSet<>(); @@ -57,7 +57,7 @@ class ProcessorBoltDelegate implements Serializable { private Multimap<String, ProcessorNode> streamToInitialProcessors; private String timestampField; - ProcessorBoltDelegate(String id, DirectedGraph<Node, Edge> graph, List<ProcessorNode> nodes) { + ProcessorBoltDelegate(String id, Graph<Node, Edge> graph, List<ProcessorNode> nodes) { this.id = id; this.graph = graph; this.nodes = new ArrayList<>(nodes); @@ -79,7 +79,7 @@ class ProcessorBoltDelegate implements Serializable { this.topoConf = topoConf; topologyContext = context; outputCollector = collector; - DirectedSubgraph<Node, Edge> subgraph = new DirectedSubgraph<>(graph, new HashSet<>(nodes), null); + AsSubgraph<Node, Edge> subgraph = new AsSubgraph<>(graph, new HashSet<>(nodes), null); TopologicalOrderIterator<Node, Edge> it = new TopologicalOrderIterator<>(subgraph); while (it.hasNext()) { Node node = it.next(); diff --git a/storm-client/src/jvm/org/apache/storm/streams/StatefulProcessorBolt.java b/storm-client/src/jvm/org/apache/storm/streams/StatefulProcessorBolt.java index 1e7e92235..d86008aca 100644 --- a/storm-client/src/jvm/org/apache/storm/streams/StatefulProcessorBolt.java +++ b/storm-client/src/jvm/org/apache/storm/streams/StatefulProcessorBolt.java @@ -17,7 +17,7 @@ import java.util.List; import java.util.Map; import java.util.Set; import org.apache.storm.shade.com.google.common.collect.Multimap; -import org.apache.storm.shade.org.jgrapht.DirectedGraph; +import org.apache.storm.shade.org.jgrapht.Graph; import org.apache.storm.state.KeyValueState; import org.apache.storm.streams.processors.StatefulProcessor; import org.apache.storm.streams.processors.UpdateStateByKeyProcessor; @@ -35,7 +35,7 @@ class StatefulProcessorBolt<K, V> extends BaseStatefulBolt<KeyValueState<K, V>> // can be UpdateStateByKey or StateQuery processors private final Set<StatefulProcessor<K, V>> statefulProcessors; - StatefulProcessorBolt(String boltId, DirectedGraph<Node, Edge> graph, List<ProcessorNode> nodes) { + StatefulProcessorBolt(String boltId, Graph<Node, Edge> graph, List<ProcessorNode> nodes) { delegate = new ProcessorBoltDelegate(boltId, graph, nodes); statefulProcessors = getStatefulProcessors(nodes); } diff --git a/storm-client/src/jvm/org/apache/storm/streams/StreamBuilder.java b/storm-client/src/jvm/org/apache/storm/streams/StreamBuilder.java index 45ed897e8..32192652b 100644 --- a/storm-client/src/jvm/org/apache/storm/streams/StreamBuilder.java +++ b/storm-client/src/jvm/org/apache/storm/streams/StreamBuilder.java @@ -20,7 +20,6 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.PriorityQueue; import java.util.Set; import java.util.stream.Collectors; import org.apache.storm.annotation.InterfaceStability; @@ -68,7 +67,7 @@ public class StreamBuilder { * Creates a new {@link StreamBuilder}. */ public StreamBuilder() { - graph = new DefaultDirectedGraph<>(new StreamsEdgeFactory()); + graph = new DefaultDirectedGraph<>(null, null, false); } /** @@ -158,7 +157,7 @@ public class StreamBuilder { nodeGroupingInfo.clear(); windowInfo.clear(); curGroup.clear(); - TopologicalOrderIterator<Node, Edge> iterator = new TopologicalOrderIterator<>(graph, queue()); + TopologicalOrderIterator<Node, Edge> iterator = new TopologicalOrderIterator<>(graph, priorityComparator()); TopologyBuilder topologyBuilder = new TopologyBuilder(); while (iterator.hasNext()) { Node node = iterator.next(); @@ -196,7 +195,7 @@ public class StreamBuilder { Node addNode(Node parent, Node child, String parentStreamId, int parallelism) { graph.addVertex(child); - graph.addEdge(parent, child); + graph.addEdge(parent, child, new Edge(parent, child)); child.setParallelism(parallelism); if (parent instanceof WindowNode || parent instanceof PartitionNode) { child.addParentStream(parentNode(parent), parentStreamId); @@ -236,9 +235,8 @@ public class StreamBuilder { return newChild; } - private PriorityQueue<Node> queue() { - // min-heap - return new PriorityQueue<>(new Comparator<Node>() { + private Comparator<Node> priorityComparator() { + return new Comparator<Node>() { /* * Nodes in the descending order of priority. * ProcessorNode has higher priority than partition and window nodes @@ -279,7 +277,7 @@ public class StreamBuilder { } return Integer.MAX_VALUE; } - }); + }; } private void handleProcessorNode(ProcessorNode processorNode, TopologyBuilder topologyBuilder) { diff --git a/storm-client/src/jvm/org/apache/storm/streams/StreamUtil.java b/storm-client/src/jvm/org/apache/storm/streams/StreamUtil.java index d4b1029b3..a6ca3f09a 100644 --- a/storm-client/src/jvm/org/apache/storm/streams/StreamUtil.java +++ b/storm-client/src/jvm/org/apache/storm/streams/StreamUtil.java @@ -16,12 +16,12 @@ import static org.apache.storm.streams.WindowNode.PUNCTUATION; import java.util.ArrayList; import java.util.List; -import org.apache.storm.shade.org.jgrapht.DirectedGraph; +import org.apache.storm.shade.org.jgrapht.Graph; import org.apache.storm.tuple.Fields; public class StreamUtil { @SuppressWarnings("unchecked") - public static <T> List<T> getParents(DirectedGraph<Node, Edge> graph, Node node) { + public static <T> List<T> getParents(Graph<Node, Edge> graph, Node node) { List<Edge> incoming = new ArrayList<>(graph.incomingEdgesOf(node)); List<T> ret = new ArrayList<>(); for (Edge e : incoming) { @@ -31,7 +31,7 @@ public class StreamUtil { } @SuppressWarnings("unchecked") - public static <T> List<T> getChildren(DirectedGraph<Node, Edge> graph, Node node) { + public static <T> List<T> getChildren(Graph<Node, Edge> graph, Node node) { List<Edge> outgoing = new ArrayList<>(graph.outgoingEdgesOf(node)); List<T> ret = new ArrayList<>(); for (Edge e : outgoing) { diff --git a/storm-client/src/jvm/org/apache/storm/streams/StreamsEdgeFactory.java b/storm-client/src/jvm/org/apache/storm/streams/StreamsEdgeFactory.java deleted file mode 100644 index ad2d6d508..000000000 --- a/storm-client/src/jvm/org/apache/storm/streams/StreamsEdgeFactory.java +++ /dev/null @@ -1,23 +0,0 @@ -/** - * 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.streams; - -import java.io.Serializable; -import org.apache.storm.shade.org.jgrapht.EdgeFactory; - -class StreamsEdgeFactory implements EdgeFactory<Node, Edge>, Serializable { - @Override - public Edge createEdge(Node sourceVertex, Node targetVertex) { - return new Edge(sourceVertex, targetVertex); - } -} diff --git a/storm-client/src/jvm/org/apache/storm/streams/WindowedProcessorBolt.java b/storm-client/src/jvm/org/apache/storm/streams/WindowedProcessorBolt.java index eb5e09bcb..1a6df0524 100644 --- a/storm-client/src/jvm/org/apache/storm/streams/WindowedProcessorBolt.java +++ b/storm-client/src/jvm/org/apache/storm/streams/WindowedProcessorBolt.java @@ -18,7 +18,7 @@ import java.util.Date; import java.util.List; import java.util.Map; import org.apache.storm.shade.com.google.common.collect.Multimap; -import org.apache.storm.shade.org.jgrapht.DirectedGraph; +import org.apache.storm.shade.org.jgrapht.Graph; import org.apache.storm.streams.windowing.SlidingWindows; import org.apache.storm.streams.windowing.TumblingWindows; import org.apache.storm.streams.windowing.Window; @@ -39,7 +39,7 @@ class WindowedProcessorBolt extends BaseWindowedBolt implements StreamBolt { private final ProcessorBoltDelegate delegate; private final Window<?, ?> window; - WindowedProcessorBolt(String id, DirectedGraph<Node, Edge> graph, + WindowedProcessorBolt(String id, Graph<Node, Edge> graph, List<ProcessorNode> nodes, Window<?, ?> window) { delegate = new ProcessorBoltDelegate(id, graph, nodes); diff --git a/storm-client/src/jvm/org/apache/storm/trident/TridentTopology.java b/storm-client/src/jvm/org/apache/storm/trident/TridentTopology.java index cfeab4b2e..98924b174 100644 --- a/storm-client/src/jvm/org/apache/storm/trident/TridentTopology.java +++ b/storm-client/src/jvm/org/apache/storm/trident/TridentTopology.java @@ -38,9 +38,8 @@ import org.apache.storm.generated.Grouping; import org.apache.storm.generated.SharedMemory; import org.apache.storm.generated.StormTopology; import org.apache.storm.grouping.CustomStreamGrouping; -import org.apache.storm.shade.org.jgrapht.DirectedGraph; -import org.apache.storm.shade.org.jgrapht.UndirectedGraph; -import org.apache.storm.shade.org.jgrapht.alg.ConnectivityInspector; +import org.apache.storm.shade.org.jgrapht.Graph; +import org.apache.storm.shade.org.jgrapht.alg.connectivity.ConnectivityInspector; import org.apache.storm.shade.org.jgrapht.graph.DefaultDirectedGraph; import org.apache.storm.shade.org.jgrapht.graph.Pseudograph; import org.apache.storm.topology.BoltDeclarer; @@ -103,7 +102,7 @@ public class TridentTopology { Map<String, Number> masterCoordResources = new HashMap<>(); public TridentTopology() { - this(new DefaultDirectedGraph<Node, IndexedEdge>(new ErrorEdgeFactory()), + this(new DefaultDirectedGraph<Node, IndexedEdge>(null, new ErrorEdgeFactory(), false), new LinkedHashMap<String, List<Node>>(), new UniqueIdGen()); } @@ -333,9 +332,9 @@ public class TridentTopology { return ret; } - private static Map<Group, Integer> getGroupParallelisms(DirectedGraph<Node, IndexedEdge> graph, GraphGrouper grouper, + private static Map<Group, Integer> getGroupParallelisms(Graph<Node, IndexedEdge> graph, GraphGrouper grouper, Collection<Group> groups) { - UndirectedGraph<Group, Object> equivs = new Pseudograph<>(Object.class); + Graph<Group, Object> equivs = new Pseudograph<>(Object.class); for (Group g : groups) { equivs.addVertex(g); } @@ -440,7 +439,7 @@ public class TridentTopology { return false; } - private static void addEdge(DirectedGraph g, Object source, Object target, int index) { + private static void addEdge(Graph g, Object source, Object target, int index) { g.addEdge(source, target, new IndexedEdge(source, target, index)); } diff --git a/storm-client/src/jvm/org/apache/storm/trident/graph/GraphGrouper.java b/storm-client/src/jvm/org/apache/storm/trident/graph/GraphGrouper.java index 88ccfeb41..3a890b70c 100644 --- a/storm-client/src/jvm/org/apache/storm/trident/graph/GraphGrouper.java +++ b/storm-client/src/jvm/org/apache/storm/trident/graph/GraphGrouper.java @@ -18,16 +18,16 @@ import java.util.HashSet; import java.util.LinkedHashSet; import java.util.Map; import java.util.Set; -import org.apache.storm.shade.org.jgrapht.DirectedGraph; +import org.apache.storm.shade.org.jgrapht.Graph; import org.apache.storm.trident.planner.Node; import org.apache.storm.trident.util.IndexedEdge; public class GraphGrouper { - final DirectedGraph<Node, IndexedEdge> graph; + final Graph<Node, IndexedEdge> graph; final Set<Group> currGroups; final Map<Node, Group> groupIndex = new HashMap<>(); - public GraphGrouper(DirectedGraph<Node, IndexedEdge> graph, Collection<Group> initialGroups) { + public GraphGrouper(Graph<Node, IndexedEdge> graph, Collection<Group> initialGroups) { this.graph = graph; this.currGroups = new LinkedHashSet<>(initialGroups); reindex(); diff --git a/storm-client/src/jvm/org/apache/storm/trident/graph/Group.java b/storm-client/src/jvm/org/apache/storm/trident/graph/Group.java index 456f1a0a6..8dc359f39 100644 --- a/storm-client/src/jvm/org/apache/storm/trident/graph/Group.java +++ b/storm-client/src/jvm/org/apache/storm/trident/graph/Group.java @@ -20,22 +20,22 @@ import java.util.Map; import java.util.Set; import java.util.UUID; import org.apache.storm.generated.SharedMemory; -import org.apache.storm.shade.org.jgrapht.DirectedGraph; +import org.apache.storm.shade.org.jgrapht.Graph; import org.apache.storm.trident.planner.Node; import org.apache.storm.trident.util.IndexedEdge; import org.apache.storm.trident.util.TridentUtils; public class Group { public final Set<Node> nodes = new HashSet<>(); - private final DirectedGraph<Node, IndexedEdge> graph; + private final Graph<Node, IndexedEdge> graph; private final String id = UUID.randomUUID().toString(); - public Group(DirectedGraph graph, List<Node> nodes) { + public Group(Graph graph, List<Node> nodes) { this.graph = graph; this.nodes.addAll(nodes); } - public Group(DirectedGraph graph, Node n) { + public Group(Graph graph, Node n) { this(graph, Arrays.asList(n)); } diff --git a/storm-client/src/jvm/org/apache/storm/trident/planner/SubtopologyBolt.java b/storm-client/src/jvm/org/apache/storm/trident/planner/SubtopologyBolt.java index 6afd99cca..59b3f20b0 100644 --- a/storm-client/src/jvm/org/apache/storm/trident/planner/SubtopologyBolt.java +++ b/storm-client/src/jvm/org/apache/storm/trident/planner/SubtopologyBolt.java @@ -20,9 +20,9 @@ import java.util.Map; import java.util.Set; import org.apache.storm.coordination.BatchOutputCollector; import org.apache.storm.generated.GlobalStreamId; -import org.apache.storm.shade.org.jgrapht.DirectedGraph; +import org.apache.storm.shade.org.jgrapht.Graph; +import org.apache.storm.shade.org.jgrapht.graph.AsSubgraph; import org.apache.storm.shade.org.jgrapht.graph.DefaultDirectedGraph; -import org.apache.storm.shade.org.jgrapht.graph.DirectedSubgraph; import org.apache.storm.shade.org.jgrapht.traverse.TopologicalOrderIterator; import org.apache.storm.task.TopologyContext; import org.apache.storm.topology.OutputFieldsDeclarer; @@ -45,7 +45,7 @@ import org.apache.storm.tuple.Tuple; public class SubtopologyBolt implements ITridentBatchBolt { private static final long serialVersionUID = 1475508603138688412L; @SuppressWarnings("rawtypes") - final DirectedGraph<Node, IndexedEdge> graph; + final Graph<Node, IndexedEdge> graph; final Set<Node> nodes; final Map<String, InitialReceiver> roots = new HashMap<>(); final Map<Node, Factory> outputFactories = new HashMap<>(); @@ -56,7 +56,7 @@ public class SubtopologyBolt implements ITridentBatchBolt { @SuppressWarnings({ "unchecked", "rawtypes" }) public SubtopologyBolt(DefaultDirectedGraph<Node, IndexedEdge> graph, Set<Node> nodes, Map<Node, String> batchGroups) { this.nodes = nodes; - this.graph = (DirectedGraph<Node, IndexedEdge>) graph.clone(); + this.graph = (Graph<Node, IndexedEdge>) graph.clone(); this.batchGroups = copyAndOnlyKeep(batchGroups, nodes); //Remove the unneeded entries from the graph @@ -95,8 +95,8 @@ public class SubtopologyBolt implements ITridentBatchBolt { context.setTaskData(n.stateInfo.id, s); } } - DirectedSubgraph<Node, ?> subgraph = new DirectedSubgraph<>(graph, nodes, null); - TopologicalOrderIterator<Node, ?> it = new TopologicalOrderIterator<>(subgraph); + AsSubgraph<Node, IndexedEdge> subgraph = new AsSubgraph<>(graph, nodes, null); + TopologicalOrderIterator<Node, IndexedEdge> it = new TopologicalOrderIterator<>(subgraph); int stateIndex = 0; while (it.hasNext()) { Node n = it.next(); diff --git a/storm-client/src/jvm/org/apache/storm/trident/util/ErrorEdgeFactory.java b/storm-client/src/jvm/org/apache/storm/trident/util/ErrorEdgeFactory.java index ba47cacf0..1af43912e 100644 --- a/storm-client/src/jvm/org/apache/storm/trident/util/ErrorEdgeFactory.java +++ b/storm-client/src/jvm/org/apache/storm/trident/util/ErrorEdgeFactory.java @@ -13,11 +13,11 @@ package org.apache.storm.trident.util; import java.io.Serializable; -import org.apache.storm.shade.org.jgrapht.EdgeFactory; +import java.util.function.Supplier; -public class ErrorEdgeFactory implements EdgeFactory, Serializable { +public class ErrorEdgeFactory implements Supplier<IndexedEdge>, Serializable { @Override - public Object createEdge(Object v, Object v1) { + public IndexedEdge get() { throw new RuntimeException("Edges should be made explicitly"); } } diff --git a/storm-client/src/jvm/org/apache/storm/trident/util/TridentUtils.java b/storm-client/src/jvm/org/apache/storm/trident/util/TridentUtils.java index a52e7e387..4c32017b0 100644 --- a/storm-client/src/jvm/org/apache/storm/trident/util/TridentUtils.java +++ b/storm-client/src/jvm/org/apache/storm/trident/util/TridentUtils.java @@ -19,7 +19,7 @@ import java.util.List; import java.util.Map; import java.util.Set; import org.apache.storm.generated.StreamInfo; -import org.apache.storm.shade.org.jgrapht.DirectedGraph; +import org.apache.storm.shade.org.jgrapht.Graph; import org.apache.storm.thrift.TBase; import org.apache.storm.topology.IComponent; import org.apache.storm.topology.OutputFieldsGetter; @@ -75,7 +75,7 @@ public class TridentUtils { /** * Assumes edge contains an index. */ - public static <T> List<T> getParents(DirectedGraph g, T n) { + public static <T> List<T> getParents(Graph g, T n) { List<IndexedEdge> incoming = new ArrayList(g.incomingEdgesOf(n)); Collections.sort(incoming); List<T> ret = new ArrayList(); @@ -85,7 +85,7 @@ public class TridentUtils { return ret; } - public static <T> List<T> getChildren(DirectedGraph g, T n) { + public static <T> List<T> getChildren(Graph g, T n) { List<IndexedEdge> outgoing = new ArrayList(g.outgoingEdgesOf(n)); List<T> ret = new ArrayList(); for (IndexedEdge e : outgoing) { @@ -95,7 +95,7 @@ public class TridentUtils { } - public static <T> T getParent(DirectedGraph g, T n) { + public static <T> T getParent(Graph g, T n) { List<T> parents = getParents(g, n); if (parents.size() != 1) { throw new RuntimeException("Expected a single parent"); diff --git a/storm-client/test/jvm/org/apache/storm/streams/ProcessorBoltTest.java b/storm-client/test/jvm/org/apache/storm/streams/ProcessorBoltTest.java index 977c1bc79..8cd97e71f 100644 --- a/storm-client/test/jvm/org/apache/storm/streams/ProcessorBoltTest.java +++ b/storm-client/test/jvm/org/apache/storm/streams/ProcessorBoltTest.java @@ -20,7 +20,7 @@ import java.util.Set; import org.apache.storm.generated.GlobalStreamId; import org.apache.storm.generated.Grouping; import org.apache.storm.shade.com.google.common.collect.Multimap; -import org.apache.storm.shade.org.jgrapht.DirectedGraph; +import org.apache.storm.shade.org.jgrapht.Graph; import org.apache.storm.shade.org.jgrapht.graph.DefaultDirectedGraph; import org.apache.storm.streams.operations.aggregators.LongSum; import org.apache.storm.streams.processors.AggregateProcessor; @@ -51,7 +51,7 @@ public class ProcessorBoltTest { Tuple mockTuple3; Tuple punctuation; Multimap<String, ProcessorNode> mockStreamToProcessors; - DirectedGraph<Node, Edge> graph; + Graph<Node, Edge> graph; @BeforeEach public void setUp() throws Exception { @@ -64,7 +64,7 @@ public class ProcessorBoltTest { punctuation = Mockito.mock(Tuple.class); setUpPunctuation(punctuation); mockStreamToProcessors = Mockito.mock(Multimap.class); - graph = new DefaultDirectedGraph(new StreamsEdgeFactory()); + graph = new DefaultDirectedGraph<>(null, null, false); } diff --git a/storm-client/test/jvm/org/apache/storm/streams/StatefulProcessorBoltTest.java b/storm-client/test/jvm/org/apache/storm/streams/StatefulProcessorBoltTest.java index af74a435a..2dd14cc6c 100644 --- a/storm-client/test/jvm/org/apache/storm/streams/StatefulProcessorBoltTest.java +++ b/storm-client/test/jvm/org/apache/storm/streams/StatefulProcessorBoltTest.java @@ -16,7 +16,7 @@ import java.util.Collection; import java.util.Collections; import java.util.HashMap; import org.apache.storm.shade.com.google.common.collect.Multimap; -import org.apache.storm.shade.org.jgrapht.DirectedGraph; +import org.apache.storm.shade.org.jgrapht.Graph; import org.apache.storm.shade.org.jgrapht.graph.DefaultDirectedGraph; import org.apache.storm.state.KeyValueState; import org.apache.storm.streams.operations.StateUpdater; @@ -43,7 +43,7 @@ public class StatefulProcessorBoltTest { OutputCollector mockOutputCollector; StatefulProcessorBolt<String, Long> bolt; Tuple mockTuple1; - DirectedGraph<Node, Edge> graph; + Graph<Node, Edge> graph; Multimap<String, ProcessorNode> mockStreamToProcessors; KeyValueState<String, Long> mockKeyValueState; @@ -86,7 +86,7 @@ public class StatefulProcessorBoltTest { ProcessorNode node = new ProcessorNode(processor, "outputstream", new Fields("value")); node.setEmitsPair(true); Mockito.when(mockStreamToProcessors.get(Mockito.anyString())).thenReturn(Collections.singletonList(node)); - graph = new DefaultDirectedGraph(new StreamsEdgeFactory()); + graph = new DefaultDirectedGraph<>(null, null, false); graph.addVertex(node); bolt = new StatefulProcessorBolt<>("bolt1", graph, Collections.singletonList(node)); bolt.setStreamToInitialProcessors(mockStreamToProcessors); diff --git a/storm-client/test/jvm/org/apache/storm/streams/WindowedProcessorBoltTest.java b/storm-client/test/jvm/org/apache/storm/streams/WindowedProcessorBoltTest.java index 994de576e..efc76e021 100644 --- a/storm-client/test/jvm/org/apache/storm/streams/WindowedProcessorBoltTest.java +++ b/storm-client/test/jvm/org/apache/storm/streams/WindowedProcessorBoltTest.java @@ -16,7 +16,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import org.apache.storm.shade.com.google.common.collect.Multimap; -import org.apache.storm.shade.org.jgrapht.DirectedGraph; +import org.apache.storm.shade.org.jgrapht.Graph; import org.apache.storm.shade.org.jgrapht.graph.DefaultDirectedGraph; import org.apache.storm.streams.operations.aggregators.Count; import org.apache.storm.streams.processors.AggregateProcessor; @@ -47,7 +47,7 @@ public class WindowedProcessorBoltTest { Tuple mockTuple1; Tuple mockTuple2; Tuple mockTuple3; - DirectedGraph<Node, Edge> graph; + Graph<Node, Edge> graph; Multimap<String, ProcessorNode> mockStreamToProcessors; @BeforeEach @@ -80,7 +80,7 @@ public class WindowedProcessorBoltTest { node.setWindowed(true); Mockito.when(mockStreamToProcessors.get(Mockito.anyString())).thenReturn(Collections.singletonList(node)); Mockito.when(mockStreamToProcessors.keySet()).thenReturn(Collections.singleton("inputstream")); - graph = new DefaultDirectedGraph<>(new StreamsEdgeFactory()); + graph = new DefaultDirectedGraph<>(null, null, false); graph.addVertex(node); bolt = new WindowedProcessorBolt("bolt1", graph, Collections.singletonList(node), window); bolt.setStreamToInitialProcessors(mockStreamToProcessors);
