Ethanlm commented on a change in pull request #3322:
URL: https://github.com/apache/storm/pull/3322#discussion_r480389623
##########
File path: storm-client/test/jvm/org/apache/storm/utils/UtilsTest.java
##########
@@ -237,4 +247,167 @@ public void checkVersionInfo() {
assertNotNull(found);
assertEquals(key, found.getVersion());
}
+
+ @Test
+ public void testFindComponentCycles() {
+ class CycleDetectionScenario {
+ final String testName;
+ final String testDescription;
+ final StormTopology topology;
+ final int expectedCycles;
+
+ CycleDetectionScenario() {
+ testName = "dummy";
+ testDescription = "dummy test";
+ topology = null;
+ expectedCycles = 0;
+ }
+
+ CycleDetectionScenario(String testName, String testDescription,
StormTopology topology, int expectedCycles) {
+ this.testName = testName.replace(' ', '-');
+ this.testDescription = testDescription;
+ this.topology = topology;
+ this.expectedCycles = expectedCycles;
+ }
+
+ public List<CycleDetectionScenario> createTestScenarios() {
+ List<CycleDetectionScenario> ret = new ArrayList<>();
+ int testNo = 0;
+ CycleDetectionScenario s;
+ TopologyBuilder tb;
+
+ // Base case
+ {
+ testNo++;
+ tb = new TopologyBuilder();
+ tb.setSpout("spout1", new TestWordSpout(), 10);
+ tb.setBolt("bolt1", new TestWordCounter(),
10).shuffleGrouping("spout1");
+ tb.setBolt("bolt2", new TestWordCounter(),
10).shuffleGrouping("spout1");
+ tb.setBolt("bolt11", new TestWordCounter(),
10).shuffleGrouping("bolt1");
+ tb.setBolt("bolt12", new TestWordCounter(),
10).shuffleGrouping("bolt1");
+ tb.setBolt("bolt21", new TestWordCounter(),
10).shuffleGrouping("bolt2");
+ tb.setBolt("bolt22", new TestWordCounter(),
10).shuffleGrouping("bolt2");
+ s = new CycleDetectionScenario(String.format("(%d) Base",
testNo),
+ "Three level component hierarchy with no loops",
+ tb.createTopology(),
+ 0);
+ ret.add(s);
+ }
+
+ // single loop with one bolt
+ {
+ testNo++;
+ tb = new TopologyBuilder();
+ tb.setSpout("spout1", new TestWordSpout(), 10);
+ tb.setBolt("bolt1", new TestWordCounter(),
10).shuffleGrouping("spout1");
+ tb.setBolt("bolt2", new TestWordCounter(),
10).shuffleGrouping("spout1");
+ tb.setBolt("bolt11", new TestWordCounter(),
10).shuffleGrouping("bolt1");
+ tb.setBolt("bolt12", new TestWordCounter(),
10).shuffleGrouping("bolt1");
+ tb.setBolt("bolt21", new TestWordCounter(),
10).shuffleGrouping("bolt2");
+ tb.setBolt("bolt22", new TestWordCounter(),
10).shuffleGrouping("bolt2");
+ // loop bolt 3 (also connect bolt3 to spout 1)
+ tb.setBolt("bolt3", new TestWordCounter(),
10).shuffleGrouping("spout1").shuffleGrouping("bolt3");
+ ret.add(new CycleDetectionScenario(String.format("(%d) One
Loop", testNo),
+ "Four level component hierarchy with 1 cycle in
bolt3",
Review comment:
Should this be "Three Level" component hierarchy?
##########
File path: storm-client/src/jvm/org/apache/storm/utils/Utils.java
##########
@@ -1907,4 +1909,117 @@ private void readArchive(ZipFile zipFile) throws
IOException {
}
}
}
+
+ /**
+ * Create a map of forward edges for bolts in a topology. Note that spouts
can be source but not a target in
+ * the edge. The mapping contains ids of spouts and bolts.
+ *
+ * @param topology StormTopology to examine.
+ * @return a map with entry for each SpoutId/BoltId to a set of outbound
edges of BoltIds.
+ */
+ private static Map<String, Set<String>>
getStormTopologyForwardGraph(StormTopology topology) {
+ Map<String, Set<String>> edgesOut = new HashMap<>();
+
+ if (topology.get_bolts() != null) {
+ topology.get_bolts().entrySet().forEach(entry -> {
+ if (!Utils.isSystemId(entry.getKey())) {
+ entry.getValue().get_common().get_inputs().forEach((k, v)
-> {
+ edgesOut.computeIfAbsent(k.get_componentId(), x -> new
HashSet<>()).add(entry.getKey());
+ });
+ }
+ });
+ }
+ return edgesOut;
+ }
+
+ /**
+ * Use recursive descent to detect cycles. This is a Depth First
recursion. Component Cycle is recorded when encountered.
+ * In addition, the last link in the cycle is removed to avoid
re-detecting same cycle/subcycle.
+ *
+ * @param stack used for recursion.
+ * @param edgesOut outbound edge connections, modified when cycle is
detected.
+ * @param seen keeps track of component ids that have already been seen.
+ * @param cycles list of cycles seen so far.
+ */
+ private static void findComponentCyclesRecursion(
+ Stack<String> stack, Map<String, Set<String>> edgesOut,
Set<String> seen, List<List<String>> cycles) {
+ if (stack.isEmpty()) {
+ return;
+ }
+ String compId1 = stack.peek();
+ if (!edgesOut.containsKey(compId1) || edgesOut.get(compId1).isEmpty())
{
+ stack.pop();
+ return;
+ }
+ Set<String> children = new HashSet<>(edgesOut.get(compId1));
+ for (String compId2: children) {
+ if (seen.contains(compId2)) {
+ // cycle/diamond detected
+ List<String> possibleCycle = new ArrayList<>();
+ if (compId1.equals(compId2)) {
+ possibleCycle.add(compId2);
+ } else if (edgesOut.get(compId2).contains(compId1)) {
Review comment:
With a slightly modified test case,
```
tb.setSpout("spout1", new TestWordSpout(), 10);
tb.setSpout("spout2", new TestWordSpout(), 10);
tb.setBolt("bolt1", new TestWordCounter(),
10).shuffleGrouping("spout1").shuffleGrouping("bolt4").shuffleGrouping("bolt2");
tb.setBolt("bolt2", new TestWordCounter(),
10).shuffleGrouping("bolt1");
tb.setBolt("bolt3", new TestWordCounter(),
10).shuffleGrouping("bolt2").shuffleGrouping("bolt4");
tb.setBolt("bolt4", new TestWordCounter(),
10).shuffleGrouping("spout2");
```
I am seeing
```
java.lang.NullPointerException
at
org.apache.storm.utils.Utils.findComponentCyclesRecursion(Utils.java:1961)
at
org.apache.storm.utils.Utils.findComponentCyclesRecursion(Utils.java:1981)
```
I am reading the code to try to figure out why
##########
File path: storm-client/test/jvm/org/apache/storm/utils/UtilsTest.java
##########
@@ -237,4 +247,167 @@ public void checkVersionInfo() {
assertNotNull(found);
assertEquals(key, found.getVersion());
}
+
+ @Test
+ public void testFindComponentCycles() {
+ class CycleDetectionScenario {
+ final String testName;
+ final String testDescription;
+ final StormTopology topology;
+ final int expectedCycles;
+
+ CycleDetectionScenario() {
+ testName = "dummy";
+ testDescription = "dummy test";
+ topology = null;
+ expectedCycles = 0;
+ }
+
+ CycleDetectionScenario(String testName, String testDescription,
StormTopology topology, int expectedCycles) {
+ this.testName = testName.replace(' ', '-');
+ this.testDescription = testDescription;
+ this.topology = topology;
+ this.expectedCycles = expectedCycles;
+ }
+
+ public List<CycleDetectionScenario> createTestScenarios() {
+ List<CycleDetectionScenario> ret = new ArrayList<>();
+ int testNo = 0;
+ CycleDetectionScenario s;
+ TopologyBuilder tb;
+
+ // Base case
+ {
+ testNo++;
+ tb = new TopologyBuilder();
+ tb.setSpout("spout1", new TestWordSpout(), 10);
+ tb.setBolt("bolt1", new TestWordCounter(),
10).shuffleGrouping("spout1");
+ tb.setBolt("bolt2", new TestWordCounter(),
10).shuffleGrouping("spout1");
+ tb.setBolt("bolt11", new TestWordCounter(),
10).shuffleGrouping("bolt1");
+ tb.setBolt("bolt12", new TestWordCounter(),
10).shuffleGrouping("bolt1");
+ tb.setBolt("bolt21", new TestWordCounter(),
10).shuffleGrouping("bolt2");
+ tb.setBolt("bolt22", new TestWordCounter(),
10).shuffleGrouping("bolt2");
+ s = new CycleDetectionScenario(String.format("(%d) Base",
testNo),
+ "Three level component hierarchy with no loops",
+ tb.createTopology(),
+ 0);
+ ret.add(s);
+ }
+
+ // single loop with one bolt
+ {
+ testNo++;
+ tb = new TopologyBuilder();
+ tb.setSpout("spout1", new TestWordSpout(), 10);
+ tb.setBolt("bolt1", new TestWordCounter(),
10).shuffleGrouping("spout1");
+ tb.setBolt("bolt2", new TestWordCounter(),
10).shuffleGrouping("spout1");
+ tb.setBolt("bolt11", new TestWordCounter(),
10).shuffleGrouping("bolt1");
+ tb.setBolt("bolt12", new TestWordCounter(),
10).shuffleGrouping("bolt1");
+ tb.setBolt("bolt21", new TestWordCounter(),
10).shuffleGrouping("bolt2");
+ tb.setBolt("bolt22", new TestWordCounter(),
10).shuffleGrouping("bolt2");
+ // loop bolt 3 (also connect bolt3 to spout 1)
+ tb.setBolt("bolt3", new TestWordCounter(),
10).shuffleGrouping("spout1").shuffleGrouping("bolt3");
+ ret.add(new CycleDetectionScenario(String.format("(%d) One
Loop", testNo),
+ "Four level component hierarchy with 1 cycle in
bolt3",
+ tb.createTopology(),
+ 1));
+ }
+
+ // single loop with three bolts
+ {
+ testNo++;
+ tb = new TopologyBuilder();
+ tb.setSpout("spout1", new TestWordSpout(), 10);
+ tb.setBolt("bolt1", new TestWordCounter(),
10).shuffleGrouping("spout1");
+ tb.setBolt("bolt2", new TestWordCounter(),
10).shuffleGrouping("spout1");
+ tb.setBolt("bolt11", new TestWordCounter(),
10).shuffleGrouping("bolt1");
+ tb.setBolt("bolt12", new TestWordCounter(),
10).shuffleGrouping("bolt1");
+ tb.setBolt("bolt21", new TestWordCounter(),
10).shuffleGrouping("bolt2");
+ tb.setBolt("bolt22", new TestWordCounter(),
10).shuffleGrouping("bolt2");
+ // loop bolt 3 -> 4 -> 5 -> 3 (also connect bolt3 to
spout1)
+ tb.setBolt("bolt3", new TestWordCounter(),
10).shuffleGrouping("spout1").shuffleGrouping("bolt5");
+ tb.setBolt("bolt4", new TestWordCounter(),
10).shuffleGrouping("bolt3");
+ tb.setBolt("bolt5", new TestWordCounter(),
10).shuffleGrouping("bolt4");
+ ret.add(new CycleDetectionScenario(String.format("(%d) One
Loop", testNo),
+ "Four level component hierarchy with 1 cycle in
bolt3,bolt4,bolt5",
+ tb.createTopology(),
+ 1));
+ }
+
+ // two loops with three bolts, and one bolt
+ {
+ testNo++;
+ tb = new TopologyBuilder();
+ tb.setSpout("spout1", new TestWordSpout(), 10);
+ tb.setBolt("bolt1", new TestWordCounter(),
10).shuffleGrouping("spout1");
+ tb.setBolt("bolt2", new TestWordCounter(),
10).shuffleGrouping("spout1");
+ tb.setBolt("bolt11", new TestWordCounter(),
10).shuffleGrouping("bolt1");
+ tb.setBolt("bolt12", new TestWordCounter(),
10).shuffleGrouping("bolt1");
+ tb.setBolt("bolt21", new TestWordCounter(),
10).shuffleGrouping("bolt2");
+ tb.setBolt("bolt22", new TestWordCounter(),
10).shuffleGrouping("bolt2");
+ // loop bolt 3 -> 4 -> 5 -> 3 (also connect bolt3 to
spout1)
+ tb.setBolt("bolt3", new TestWordCounter(),
10).shuffleGrouping("spout1").shuffleGrouping("bolt5");
+ tb.setBolt("bolt4", new TestWordCounter(),
10).shuffleGrouping("bolt3");
+ tb.setBolt("bolt5", new TestWordCounter(),
10).shuffleGrouping("bolt4");
+ // loop bolt 6 (also connect bolt6 to spout 1)
+ tb.setBolt("bolt6", new TestWordCounter(),
10).shuffleGrouping("spout1").shuffleGrouping("bolt6");
+ ret.add(new CycleDetectionScenario(String.format("(%d) Two
Loops", testNo),
+ "Four level component hierarchy with 2 cycles in
bolt3,bolt4,bolt5 and bolt6",
+ tb.createTopology(),
+ 2));
+ }
+
+ // complex cycle
+ {
+ // (S1 -> B1 -> B2 -> B3 -> B4 <- S2), (B4 -> B3), (B4 ->
B2)
Review comment:
The description seems incorrect.
It should be
```
// (S1 -> B1 -> B2 -> B3 -> B4 <- S2), (B4 -> B3), (B4 -> B1)
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
