[STORM-3650] Add topology config to attempt to schedule system component evenly throughout workers #3283
Conversation
|
Travis has a ExecutorTransferMultiThreadingTest failed. Re-run the test locally and it passed. |
| @@ -1695,10 +1695,10 @@ public HeartbeatCache getHeartbeatsCache() { | |||
|
|
|||
| private Set<List<Integer>> getOrUpdateExecutors(String topoId, StormBase base, Map<String, Object> topoConf, | |||
| StormTopology topology) | |||
| throws IOException, AuthorizationException, InvalidTopologyException, KeyNotFoundException { | |||
| throws InvalidTopologyException { | |||
There was a problem hiding this comment.
Can the Nimbus cleanup of parameter (topoId) and throws clause be part of a separate Jira?
| @@ -84,6 +89,8 @@ public TopologyDetails(String topologyId, Map<String, Object> topologyConf, Stor | |||
| initConfigs(); | |||
| this.launchTime = launchTime; | |||
| this.topoName = (String) topologyConf.get(Config.TOPOLOGY_NAME); | |||
| this.distributeSysCompEvenly | |||
There was a problem hiding this comment.
This property is more useful in Scheduler code where is can be used to sort executors in a manner conducive to even distribution.
| * with the list of its parents, children and execs assigned to that component. | ||
| * | ||
| * @return a map of components | ||
| */ | ||
| public Map<String, Component> getComponents() { | ||
| public Map<String, Component> getTopoComponents() { | ||
| if (topologyComponentsMap == null) { |
There was a problem hiding this comment.
TopologyDetails gets instantiated a large number of times. It was my observation that adding members or such cacheing in TopologyDetails was counterproductive. topologyComponentsMap and its sister cache allComponentsMap should not be added unless we actually gain from cacheing.
| * with the list of its parents, children and execs assigned to that component. | ||
| * | ||
| * @return a map of components | ||
| */ | ||
| public Map<String, Component> getComponents() { | ||
| public Map<String, Component> getTopoComponents() { |
There was a problem hiding this comment.
Can this method retain the old name? And the getAllComponents be an overloaded getComponents()?
| @@ -539,30 +541,34 @@ protected String nodeToRack(RasNode node) { | |||
| * @return a list of executors in sorted order | |||
| */ | |||
| private List<ExecutorDetails> orderExecutorsDefault( | |||
There was a problem hiding this comment.
It will be convenient if the even-distribution executor re-sorting was a single method that could be applied to executor list.
This way the upcoming refactoring of the scheduling strategies would be easier.
| @@ -40,5 +42,5 @@ | |||
| * @return returns a SchedulingResult object containing SchedulingStatus object to indicate whether scheduling is | |||
| * successful. | |||
| */ | |||
| SchedulingResult schedule(Cluster schedulingState, TopologyDetails td); | |||
| SchedulingResult schedule(Cluster schedulingState, TopologyDetails td) throws InvalidTopologyException; | |||
There was a problem hiding this comment.
Why is InvalidTopologyException being thrown in schedule.
…onal acker-per-worker config
What is the purpose of the change
Add topology config for RAS scheduler to attempt to schedule system components/executors across all workers evenly instead of scheduling them all together after having finished all topology components.
How was the change tested
storm jar /home/y/lib64/jars/storm-starter.jar org.apache.storm.starter.WordCountTopology wc1 -c topology..system.components.even.distribution=true -c topology.acker.executors=8 -c topology.component.resources.onheap.memory.mb=100 -c topology.acker.resources.onheap.memory.mb=150 -c topology.workers=4 -c experimental.topology.ras.order.executors.by.proximity.needs=true -c topology.worker.max.heap.size.mb=1000storm jar /home/y/lib64/jars/storm-starter.jar org.apache.storm.starter.WordCountTopology wc-no-even-distribution -c topology.acker.executors=8 -c topology.component.resources.onheap.memory.mb=100 -c topology.acker.resources.onheap.memory.mb=150 -c topology.workers=4 -c experimental.topology.ras.order.executors.by.proximity.needs=true -c topology.worker.max.heap.size.mb=1000