-
Notifications
You must be signed in to change notification settings - Fork 4.1k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[STORM-3685] Detect and prevent cycles when Topology is submitted. #3322
Conversation
Close/Reopen for the rebuild - failures due to license file failures for JDK 11? |
close/reopen to rebuild. |
close/reopen to rebuild. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I have some minor comments (including previous comments above). Otherwise looks good to me.
@@ -233,7 +235,7 @@ public static void submitTopologyAs(String name, Map<String, Object> topoConf, S | |||
conf.putAll(topoConf); | |||
topoConf.putAll(prepareZookeeperAuthentication(conf)); | |||
|
|||
validateConfs(conf, topology); | |||
validateConfs(conf, topology, name); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I feel like this is not the right place to detect cycles. We should probably put it in another function, like
StormCommon.validateBasic
or create a new method.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Created new method StormCommon.validateCycleFree
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Moving it to Utils is also okay, like
Utils.validateTopologyBlobStoreMap
Utils.validateTopologyName
etc.
Probably makes more sense since findComponentCycles
is implemented inside Utils.
And we could make Utils.findComponentCycles
"@VisibleForTesting". So the public method will be only Utils. validateCycleFree
and the internal implementation can be changed if we'd like in the future.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@bipinprasad
Sorry for not being clear at the first comment. What do you think about moving it to Utils?^ Trying to not expose too many public methods. Also StormCommon is org.apache.storm.daemon.StormCommon
. StormSubmitter is completely on client side. I feel like Utils is a better place for this.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ok. Will switch.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Moved from StormCommons to Utils.
…topologies for cycle detection.
…lt connections counts. Fix topo comments.
…n StormSubmitter.submitTopologyAs().
@@ -51,6 +52,7 @@ | |||
import org.apache.storm.validation.ConfigValidation; | |||
import org.slf4j.Logger; | |||
import org.slf4j.LoggerFactory; | |||
import sun.net.ProgressListener; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Is this used anywhere?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
No. Will remove this.
StormCommon.validateCycleFree(topology, name); | ||
} catch (InvalidTopologyException ex) { | ||
LOG.warn(ex.get_msg(), ex); | ||
System.out.println(ex.get_msg()); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We should remove this
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Did we not want to warn the user when submitting?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LOG.warn
should be sufficient. We don't need System.out.println()
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ok. removed sysout.
try { | ||
StormCommon.validateCycleFree(topology, name); | ||
} catch (InvalidTopologyException ex) { | ||
LOG.warn(ex.get_msg(), ex); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Will ex
already contains ex.get_msg()
? If so, we will see repeating messages, which should be avoid.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The constructor seems to imply that the message will not be in the Exception, since the proper super method is not called.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Do you have an output example that I can take a look?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
InvalidTopologyException constructor does not call Exception.super(msg): https://github.com/apache/storm/blob/master/storm-client/src/jvm/org/apache/storm/generated/InvalidTopologyException.java#L111
hence, its msg variable is not available in the Exception.getMessage()
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Right. So the log will look like
19:22:23.981 [main] WARN o.a.s.StormSubmitter - Topology wc contains cycles in components "count,split"
org.apache.storm.generated.InvalidTopologyException: null
at org.apache.storm.daemon.StormCommon.validateCycleFree(StormCommon.java:596) ~[storm-client-2.3.0-SNAPSHOT.jar:2.3.0-SNAPSHOT]
at org.apache.storm.StormSubmitter.submitTopologyAs(StormSubmitter.java:244) [storm-client-2.3.0-SNAPSHOT.jar:2.3.0-SNAPSHOT]
at org.apache.storm.StormSubmitter.submitTopology(StormSubmitter.java:214) [storm-client-2.3.0-SNAPSHOT.jar:2.3.0-SNAPSHOT]
at org.apache.storm.StormSubmitter.submitTopology(StormSubmitter.java:177) [storm-client-2.3.0-SNAPSHOT.jar:2.3.0-SNAPSHOT]
at org.apache.storm.topology.ConfigurableTopology.submit(ConfigurableTopology.java:119) [storm-client-2.3.0-SNAPSHOT.jar:2.3.0-SNAPSHOT]
at org.apache.storm.starter.WordCountTopology.run(WordCountTopology.java:58) [storm-starter-2.3.0-SNAPSHOT.jar:2.3.0-SNAPSHOT]
at org.apache.storm.topology.ConfigurableTopology.start(ConfigurableTopology.java:68) [storm-client-2.3.0-SNAPSHOT.jar:2.3.0-SNAPSHOT]
at org.apache.storm.starter.WordCountTopology.main(WordCountTopology.java:36) [storm-starter-2.3.0-SNAPSHOT.jar:2.3.0-SNAPSHOT]
We can probably use WrappedInvalidTopologyException inside validateCycleFree
method so the stacktrace will make more sense. What do you think?
It will be like
org.apache.storm.utils.WrappedInvalidTopologyException: Topology wc2 contains cycles in components "count,split"
at org.apache.storm.daemon.StormCommon.validateCycleFree(StormCommon.java:596) ~[storm-client-2.3.0-SNAPSHOT.jar:2.3.0-SNAPSHOT]
.....
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
That is better than a scary "null" in the stack trace.
Changed.
…fix @throws tag; remove superfluous sysout..
…idTopologyException.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
+1
Thank you so much for addressing all the review comments
What is the purpose of the change
Topology is expected to be a Directed Acyclic Graph. Cycles in component flow can cause unexpected behavior, for example a deadlock when one of the loop components signals a back-pressure. Or properly detecting proximity when scheduling.
How was the change tested
New test to create topology with and without cycles and test detection.