diff --git a/DEPENDENCY-LICENSES b/DEPENDENCY-LICENSES index 3ab6d7872e3..fa6fd3f41ba 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 cfd9ddbe8fe..d5fdee11b30 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/pom.xml b/pom.xml index 6224cf2729c..9f6ea55ca5d 100644 --- a/pom.xml +++ b/pom.xml @@ -99,7 +99,7 @@ 4.5.14 4.4.16 4.0.6 - 0.9.0 + 1.5.3 33.6.0-jre 1.1.1 2.0.74.Final 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 559b40aca93..5c8fc13cf78 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 @@ 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 @@ class ProcessorBolt extends BaseRichBolt implements StreamBolt { private final ProcessorBoltDelegate delegate; - ProcessorBolt(String id, DirectedGraph graph, List nodes) { + ProcessorBolt(String id, Graph graph, List 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 f68746f12a0..ccb6511003e 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.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 @@ class ProcessorBoltDelegate implements Serializable { private static final Logger LOG = LoggerFactory.getLogger(ProcessorBoltDelegate.class); private final String id; - private final DirectedGraph graph; + private final Graph graph; private final List nodes; private final List outgoingProcessors = new ArrayList<>(); private final Set emittingProcessorContexts = new HashSet<>(); @@ -57,7 +57,7 @@ class ProcessorBoltDelegate implements Serializable { private Multimap streamToInitialProcessors; private String timestampField; - ProcessorBoltDelegate(String id, DirectedGraph graph, List nodes) { + ProcessorBoltDelegate(String id, Graph graph, List nodes) { this.id = id; this.graph = graph; this.nodes = new ArrayList<>(nodes); @@ -79,7 +79,7 @@ void prepare(Map topoConf, TopologyContext context, OutputCollec this.topoConf = topoConf; topologyContext = context; outputCollector = collector; - DirectedSubgraph subgraph = new DirectedSubgraph<>(graph, new HashSet<>(nodes), null); + AsSubgraph subgraph = new AsSubgraph<>(graph, new HashSet<>(nodes), null); TopologicalOrderIterator 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 1e7e922353a..d86008acad7 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.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 extends BaseStatefulBolt> // can be UpdateStateByKey or StateQuery processors private final Set> statefulProcessors; - StatefulProcessorBolt(String boltId, DirectedGraph graph, List nodes) { + StatefulProcessorBolt(String boltId, Graph graph, List 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 45ed897e8a7..32192652b6c 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.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 StormTopology build() { nodeGroupingInfo.clear(); windowInfo.clear(); curGroup.clear(); - TopologicalOrderIterator iterator = new TopologicalOrderIterator<>(graph, queue()); + TopologicalOrderIterator iterator = new TopologicalOrderIterator<>(graph, priorityComparator()); TopologyBuilder topologyBuilder = new TopologyBuilder(); while (iterator.hasNext()) { Node node = iterator.next(); @@ -196,7 +195,7 @@ Node addNode(Node parent, Node child, String parentStreamId) { 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 @@ Node insert(Node parent, Node child) { return newChild; } - private PriorityQueue queue() { - // min-heap - return new PriorityQueue<>(new Comparator() { + private Comparator priorityComparator() { + return new Comparator() { /* * Nodes in the descending order of priority. * ProcessorNode has higher priority than partition and window nodes @@ -279,7 +277,7 @@ private int getPriority(Node node) { } 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 d4b1029b33f..a6ca3f09a79 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 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 List getParents(DirectedGraph graph, Node node) { + public static List getParents(Graph graph, Node node) { List incoming = new ArrayList<>(graph.incomingEdgesOf(node)); List ret = new ArrayList<>(); for (Edge e : incoming) { @@ -31,7 +31,7 @@ public static List getParents(DirectedGraph graph, Node node) } @SuppressWarnings("unchecked") - public static List getChildren(DirectedGraph graph, Node node) { + public static List getChildren(Graph graph, Node node) { List outgoing = new ArrayList<>(graph.outgoingEdgesOf(node)); List 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 ad2d6d5081c..00000000000 --- 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, 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 eb5e09bcb11..1a6df0524f5 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.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 graph, + WindowedProcessorBolt(String id, Graph graph, List 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 cfeab4b2edf..98924b174ad 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.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 masterCoordResources = new HashMap<>(); public TridentTopology() { - this(new DefaultDirectedGraph(new ErrorEdgeFactory()), + this(new DefaultDirectedGraph(null, new ErrorEdgeFactory(), false), new LinkedHashMap>(), new UniqueIdGen()); } @@ -333,9 +332,9 @@ private static Set committerBatches(Group g, Map batchGrou return ret; } - private static Map getGroupParallelisms(DirectedGraph graph, GraphGrouper grouper, + private static Map getGroupParallelisms(Graph graph, GraphGrouper grouper, Collection groups) { - UndirectedGraph equivs = new Pseudograph<>(Object.class); + Graph equivs = new Pseudograph<>(Object.class); for (Group g : groups) { equivs.addVertex(g); } @@ -440,7 +439,7 @@ private static boolean isIdentityPartition(PartitionNode n) { 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 88ccfeb41f4..3a890b70c1f 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.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 graph; + final Graph graph; final Set currGroups; final Map groupIndex = new HashMap<>(); - public GraphGrouper(DirectedGraph graph, Collection initialGroups) { + public GraphGrouper(Graph graph, Collection 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 456f1a0a62c..8dc359f3910 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.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 nodes = new HashSet<>(); - private final DirectedGraph graph; + private final Graph graph; private final String id = UUID.randomUUID().toString(); - public Group(DirectedGraph graph, List nodes) { + public Group(Graph graph, List 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 6afd99cca8e..59b3f20b0bc 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.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 @@ public class SubtopologyBolt implements ITridentBatchBolt { private static final long serialVersionUID = 1475508603138688412L; @SuppressWarnings("rawtypes") - final DirectedGraph graph; + final Graph graph; final Set nodes; final Map roots = new HashMap<>(); final Map outputFactories = new HashMap<>(); @@ -56,7 +56,7 @@ public class SubtopologyBolt implements ITridentBatchBolt { @SuppressWarnings({ "unchecked", "rawtypes" }) public SubtopologyBolt(DefaultDirectedGraph graph, Set nodes, Map batchGroups) { this.nodes = nodes; - this.graph = (DirectedGraph) graph.clone(); + this.graph = (Graph) graph.clone(); this.batchGroups = copyAndOnlyKeep(batchGroups, nodes); //Remove the unneeded entries from the graph @@ -95,8 +95,8 @@ public void prepare(Map conf, TopologyContext context, BatchOutp context.setTaskData(n.stateInfo.id, s); } } - DirectedSubgraph subgraph = new DirectedSubgraph<>(graph, nodes, null); - TopologicalOrderIterator it = new TopologicalOrderIterator<>(subgraph); + AsSubgraph subgraph = new AsSubgraph<>(graph, nodes, null); + TopologicalOrderIterator 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 ba47cacf06c..1af43912ed0 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, 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 a52e7e3870b..4c32017b09b 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.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 static Fields getSingleOutputStreamFields(IComponent component) { /** * Assumes edge contains an index. */ - public static List getParents(DirectedGraph g, T n) { + public static List getParents(Graph g, T n) { List incoming = new ArrayList(g.incomingEdgesOf(n)); Collections.sort(incoming); List ret = new ArrayList(); @@ -85,7 +85,7 @@ public static List getParents(DirectedGraph g, T n) { return ret; } - public static List getChildren(DirectedGraph g, T n) { + public static List getChildren(Graph g, T n) { List outgoing = new ArrayList(g.outgoingEdgesOf(n)); List ret = new ArrayList(); for (IndexedEdge e : outgoing) { @@ -95,7 +95,7 @@ public static List getChildren(DirectedGraph g, T n) { } - public static T getParent(DirectedGraph g, T n) { + public static T getParent(Graph g, T n) { List 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 977c1bc79ff..8cd97e71f4a 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 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 mockStreamToProcessors; - DirectedGraph graph; + Graph graph; @BeforeEach public void setUp() throws Exception { @@ -64,7 +64,7 @@ public void setUp() throws Exception { 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 af74a435a04..2dd14cc6ca6 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.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 bolt; Tuple mockTuple1; - DirectedGraph graph; + Graph graph; Multimap mockStreamToProcessors; KeyValueState mockKeyValueState; @@ -86,7 +86,7 @@ private void setUpStatefulProcessorBolt(Processor processor) { 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 994de576e99..efc76e021f7 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.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 graph; + Graph graph; Multimap mockStreamToProcessors; @BeforeEach @@ -80,7 +80,7 @@ private void setUpWindowedProcessorBolt(Processor processor, Window win 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);