You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2017/11/20 14:30:25 UTC
[1/4] storm git commit: STORM-2805: Addressed review comments and
fixed some bugs
Repository: storm
Updated Branches:
refs/heads/master 8e4a14654 -> 9d52435c5
STORM-2805: Addressed review comments and fixed some bugs
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/dcfbf274
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/dcfbf274
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/dcfbf274
Branch: refs/heads/master
Commit: dcfbf274b0b6d75051423eb81440c85b4b27beb3
Parents: 96c461d
Author: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Authored: Thu Nov 16 16:38:16 2017 -0600
Committer: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Committed: Fri Nov 17 07:32:23 2017 -0600
----------------------------------------------------------------------
storm-client/pom.xml | 2 +-
.../coordination/BatchSubtopologyBuilder.java | 21 ++--
.../storm/drpc/LinearDRPCTopologyBuilder.java | 22 ++--
.../topology/BaseConfigurationDeclarer.java | 9 --
.../ComponentConfigurationDeclarer.java | 7 --
.../apache/storm/topology/TopologyBuilder.java | 59 +++++----
.../TransactionalTopologyBuilder.java | 38 +++---
.../topology/TridentTopologyBuilder.java | 125 +++++++++----------
storm-server/pom.xml | 2 +-
.../apache/storm/scheduler/TopologyDetails.java | 3 +-
.../storm/scheduler/resource/ResourceUtils.java | 8 +-
.../TestGenericResourceAwareStrategy.java | 66 ++++++----
12 files changed, 182 insertions(+), 180 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/dcfbf274/storm-client/pom.xml
----------------------------------------------------------------------
diff --git a/storm-client/pom.xml b/storm-client/pom.xml
index 8a7efc4..6097b35 100644
--- a/storm-client/pom.xml
+++ b/storm-client/pom.xml
@@ -261,7 +261,7 @@
<!--Note - the version would be inherited-->
<configuration>
<excludes>**/generated/**</excludes>
- <maxAllowedViolations>10320</maxAllowedViolations>
+ <maxAllowedViolations>10298</maxAllowedViolations>
</configuration>
</plugin>
<plugin>
http://git-wip-us.apache.org/repos/asf/storm/blob/dcfbf274/storm-client/src/jvm/org/apache/storm/coordination/BatchSubtopologyBuilder.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/coordination/BatchSubtopologyBuilder.java b/storm-client/src/jvm/org/apache/storm/coordination/BatchSubtopologyBuilder.java
index 9b1495d..d8e5231 100644
--- a/storm-client/src/jvm/org/apache/storm/coordination/BatchSubtopologyBuilder.java
+++ b/storm-client/src/jvm/org/apache/storm/coordination/BatchSubtopologyBuilder.java
@@ -93,9 +93,7 @@ public class BatchSubtopologyBuilder {
for (InputDeclaration decl: masterBolt.declarations) {
decl.declare(declarer);
}
- if (!masterBolt.componentConf.isEmpty()) {
- declarer.addConfigurations(masterBolt.componentConf);
- }
+ declarer.addConfigurations(masterBolt.componentConf);
for (String id: bolts.keySet()) {
Component component = bolts.get(id);
Map<String, SourceArgs> coordinatedArgs = new HashMap<String, SourceArgs>();
@@ -451,16 +449,20 @@ public class BatchSubtopologyBuilder {
@Override
public BoltDeclarer addConfigurations(Map<String, Object> conf) {
- if (conf != null && !conf.isEmpty()) {
+ if (conf != null) {
component.componentConf.putAll(conf);
}
return this;
}
@Override
- public Map<String, Object> getRASConfiguration() {
- //TODO this should not be modifiable
- return component.componentConf;
+ public BoltDeclarer addResources(Map<String, Double> resources) {
+ if (resources != null) {
+ Map<String, Double> currentResources = (Map<String, Double>) component.componentConf.computeIfAbsent(
+ Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, (k) -> new HashMap<>());
+ currentResources.putAll(resources);
+ }
+ return this;
}
@Override
@@ -472,11 +474,10 @@ public class BatchSubtopologyBuilder {
@SuppressWarnings("unchecked")
@Override
public BoltDeclarer addResource(String resourceName, Number resourceValue) {
- Map<String, Double> resourcesMap = (Map<String, Double>) getRASConfiguration().get(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP);
+ Map<String, Double> resourcesMap = (Map<String, Double>) component.componentConf.computeIfAbsent(
+ Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, (k) -> new HashMap<>());
resourcesMap.put(resourceName, resourceValue.doubleValue());
-
- getRASConfiguration().put(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, resourcesMap);
return this;
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/dcfbf274/storm-client/src/jvm/org/apache/storm/drpc/LinearDRPCTopologyBuilder.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/drpc/LinearDRPCTopologyBuilder.java b/storm-client/src/jvm/org/apache/storm/drpc/LinearDRPCTopologyBuilder.java
index 51db422..6844ad0 100644
--- a/storm-client/src/jvm/org/apache/storm/drpc/LinearDRPCTopologyBuilder.java
+++ b/storm-client/src/jvm/org/apache/storm/drpc/LinearDRPCTopologyBuilder.java
@@ -401,29 +401,33 @@ public class LinearDRPCTopologyBuilder {
@Override
public LinearDRPCInputDeclarer addConfigurations(Map<String, Object> conf) {
- if (conf != null && !conf.isEmpty()) {
+ if (conf != null) {
component.componentConf.putAll(conf);
}
return this;
}
+ @Override
+ public LinearDRPCInputDeclarer addResources(Map<String, Double> resources) {
+ if (resources != null) {
+ Map<String, Double> currentResources = (Map<String, Double>) component.componentConf.computeIfAbsent(
+ Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, (k) -> new HashMap<>());
+ currentResources.putAll(resources);
+ }
+ return this;
+ }
+
@SuppressWarnings("unchecked")
@Override
public LinearDRPCInputDeclarer addResource(String resourceName, Number resourceValue) {
- Map<String, Double> resourcesMap = (Map<String, Double>) getRASConfiguration().get(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP);
+ Map<String, Double> resourcesMap = (Map<String, Double>) component.componentConf.computeIfAbsent(
+ Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, (k) -> new HashMap<>());
resourcesMap.put(resourceName, resourceValue.doubleValue());
-
- getRASConfiguration().put(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, resourcesMap);
return this;
}
@Override
- public Map getRASConfiguration() {
- return component.componentConf;
- }
-
- @Override
public LinearDRPCInputDeclarer addSharedMemory(SharedMemory request) {
component.sharedMemory.add(request);
return this;
http://git-wip-us.apache.org/repos/asf/storm/blob/dcfbf274/storm-client/src/jvm/org/apache/storm/topology/BaseConfigurationDeclarer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/topology/BaseConfigurationDeclarer.java b/storm-client/src/jvm/org/apache/storm/topology/BaseConfigurationDeclarer.java
index eac23bb..6515c46 100644
--- a/storm-client/src/jvm/org/apache/storm/topology/BaseConfigurationDeclarer.java
+++ b/storm-client/src/jvm/org/apache/storm/topology/BaseConfigurationDeclarer.java
@@ -86,13 +86,4 @@ public abstract class BaseConfigurationDeclarer<T extends ComponentConfiguration
}
return (T) this;
}
-
- @SuppressWarnings("unchecked")
- @Override
- public T addResources(Map<String, Double> resources) {
- if (resources != null) {
- return addConfiguration(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, resources);
- }
- return (T) this;
- }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/dcfbf274/storm-client/src/jvm/org/apache/storm/topology/ComponentConfigurationDeclarer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/topology/ComponentConfigurationDeclarer.java b/storm-client/src/jvm/org/apache/storm/topology/ComponentConfigurationDeclarer.java
index 8098029..ce38392 100644
--- a/storm-client/src/jvm/org/apache/storm/topology/ComponentConfigurationDeclarer.java
+++ b/storm-client/src/jvm/org/apache/storm/topology/ComponentConfigurationDeclarer.java
@@ -29,13 +29,6 @@ public interface ComponentConfigurationDeclarer<T extends ComponentConfiguration
T addConfigurations(Map<String, Object> conf);
/**
- * return the configuration.
- *
- * @return the current configuration.
- */
- Map<String, Object> getRASConfiguration();
-
- /**
* Add in a single config.
* @param config the key for the config
* @param value the value of the config
http://git-wip-us.apache.org/repos/asf/storm/blob/dcfbf274/storm-client/src/jvm/org/apache/storm/topology/TopologyBuilder.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/topology/TopologyBuilder.java b/storm-client/src/jvm/org/apache/storm/topology/TopologyBuilder.java
index 0ebdae6..f2e09b8 100644
--- a/storm-client/src/jvm/org/apache/storm/topology/TopologyBuilder.java
+++ b/storm-client/src/jvm/org/apache/storm/topology/TopologyBuilder.java
@@ -50,8 +50,6 @@ import org.json.simple.JSONValue;
import java.io.NotSerializableException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -117,7 +115,7 @@ import static org.apache.storm.utils.Utils.parseJson;
public class TopologyBuilder {
private final Map<String, IRichBolt> _bolts = new HashMap<>();
private final Map<String, IRichSpout> _spouts = new HashMap<>();
- private final Map<String, ComponentCommon> _commons = new HashMap<>();
+ private final Map<String, ComponentCommon> commons = new HashMap<>();
private final Map<String, Set<String>> _componentToSharedMemory = new HashMap<>();
private final Map<String, SharedMemory> _sharedMemory = new HashMap<>();
private boolean hasStatefulBolt = false;
@@ -543,7 +541,7 @@ public class TopologyBuilder {
}
private ComponentCommon getComponentCommon(String id, IComponent component) {
- ComponentCommon ret = new ComponentCommon(_commons.get(id));
+ ComponentCommon ret = new ComponentCommon(commons.get(id));
OutputFieldsGetter getter = new OutputFieldsGetter();
component.declareOutputFields(getter);
ret.set_streams(getter.getFieldsDeclaration());
@@ -562,43 +560,54 @@ public class TopologyBuilder {
}
Map<String, Object> conf = component.getComponentConfiguration();
if(conf!=null) common.set_json_conf(JSONValue.toJSONString(conf));
- _commons.put(id, common);
+ commons.put(id, common);
}
protected class ConfigGetter<T extends ComponentConfigurationDeclarer> extends BaseConfigurationDeclarer<T> {
- String _id;
+ String id;
public ConfigGetter(String id) {
- _id = id;
+ this.id = id;
}
@SuppressWarnings("unchecked")
@Override
public T addConfigurations(Map<String, Object> conf) {
- if(conf!=null && conf.containsKey(Config.TOPOLOGY_KRYO_REGISTER)) {
- throw new IllegalArgumentException("Cannot set serializations for a component using fluent API");
+ if (conf != null) {
+ if (conf.containsKey(Config.TOPOLOGY_KRYO_REGISTER)) {
+ throw new IllegalArgumentException("Cannot set serializations for a component using fluent API");
+ }
+ if (!conf.isEmpty()) {
+ String currConf = commons.get(id).get_json_conf();
+ commons.get(id).set_json_conf(mergeIntoJson(parseJson(currConf), conf));
+ }
}
- String currConf = _commons.get(_id).get_json_conf();
- _commons.get(_id).set_json_conf(mergeIntoJson(parseJson(currConf), conf));
return (T) this;
}
- @SuppressWarnings("unchecked")
@Override
- public T addResource(String resourceName, Number resourceValue) {
- Map<String, Double> resourcesMap = (Map<String, Double>) getRASConfiguration().getOrDefault(
- Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, new HashMap());
-
- resourcesMap.put(resourceName, resourceValue.doubleValue());
-
- getRASConfiguration().put(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, resourcesMap);
+ public T addResources(Map<String, Double> resources) {
+ if (resources != null && !resources.isEmpty()) {
+ String currConf = commons.get(id).get_json_conf();
+ Map<String, Object> conf = parseJson(currConf);
+ Map<String, Double> currentResources =
+ (Map<String, Double>) conf.computeIfAbsent(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, (k) -> new HashMap<>());
+ currentResources.putAll(resources);
+ commons.get(id).set_json_conf(JSONValue.toJSONString(currConf));
+ }
return (T) this;
}
@SuppressWarnings("unchecked")
@Override
- public Map getRASConfiguration() {
- return parseJson(_commons.get(_id).get_json_conf());
+ public T addResource(String resourceName, Number resourceValue) {
+ Map<String, Object> componentConf = parseJson(commons.get(id).get_json_conf());
+ Map<String, Double> resourcesMap = (Map<String, Double>) componentConf.computeIfAbsent(
+ Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, (k) -> new HashMap<>());
+
+ resourcesMap.put(resourceName, resourceValue.doubleValue());
+
+ return addConfiguration(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, resourcesMap);
}
@SuppressWarnings("unchecked")
@@ -609,7 +618,7 @@ public class TopologyBuilder {
throw new IllegalArgumentException("Cannot have multiple different shared memory regions with the same name");
}
_sharedMemory.put(request.get_name(), request);
- Set<String> mems = _componentToSharedMemory.computeIfAbsent(_id, (k) -> new HashSet<>());
+ Set<String> mems = _componentToSharedMemory.computeIfAbsent(id, (k) -> new HashSet<>());
mems.add(request.get_name());
return (T) this;
}
@@ -686,7 +695,7 @@ public class TopologyBuilder {
}
private BoltDeclarer grouping(String componentId, String streamId, Grouping grouping) {
- _commons.get(_boltId).put_to_inputs(new GlobalStreamId(componentId, streamId), grouping);
+ commons.get(_boltId).put_to_inputs(new GlobalStreamId(componentId, streamId), grouping);
return this;
}
@@ -717,8 +726,8 @@ public class TopologyBuilder {
}
private static String mergeIntoJson(Map into, Map newMap) {
- Map res = new HashMap(into);
- if(newMap!=null) res.putAll(newMap);
+ Map res = new HashMap<>(into);
+ res.putAll(newMap);
return JSONValue.toJSONString(res);
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/dcfbf274/storm-client/src/jvm/org/apache/storm/transactional/TransactionalTopologyBuilder.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/transactional/TransactionalTopologyBuilder.java b/storm-client/src/jvm/org/apache/storm/transactional/TransactionalTopologyBuilder.java
index 6f994ed..466bd4c 100644
--- a/storm-client/src/jvm/org/apache/storm/transactional/TransactionalTopologyBuilder.java
+++ b/storm-client/src/jvm/org/apache/storm/transactional/TransactionalTopologyBuilder.java
@@ -224,16 +224,20 @@ public class TransactionalTopologyBuilder {
private class SpoutDeclarerImpl extends BaseConfigurationDeclarer<SpoutDeclarer> implements SpoutDeclarer {
@Override
public SpoutDeclarer addConfigurations(Map<String, Object> conf) {
- if (conf != null && !conf.isEmpty()) {
+ if (conf != null) {
spoutConf.putAll(conf);
}
return this;
}
@Override
- public Map getRASConfiguration() {
- //TODO this should be imutable
- return spoutConf;
+ public SpoutDeclarerImpl addResources(Map<String, Double> resources) {
+ if (resources != null) {
+ Map<String, Double> currentResources = (Map<String, Double>) spoutConf.computeIfAbsent(
+ Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, (k) -> new HashMap<>());
+ currentResources.putAll(resources);
+ }
+ return this;
}
@Override
@@ -245,14 +249,10 @@ public class TransactionalTopologyBuilder {
@SuppressWarnings("unchecked")
@Override
public SpoutDeclarer addResource(String resourceName, Number resourceValue) {
- Map<String, Double> resourcesMap = (Map<String, Double>) getRASConfiguration().get(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP);
+ Map<String, Double> resourcesMap = (Map<String, Double>) spoutConf.computeIfAbsent(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP,
+ (k) -> new HashMap<>());
- if (resourcesMap == null) {
- resourcesMap = new HashMap<>();
- }
resourcesMap.put(resourceName, resourceValue.doubleValue());
-
- getRASConfiguration().put(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, resourcesMap);
return this;
}
}
@@ -552,18 +552,21 @@ public class TransactionalTopologyBuilder {
@Override
public BoltDeclarer addConfigurations(Map<String, Object> conf) {
- if (conf != null && !conf.isEmpty()) {
+ if (conf != null) {
component.componentConf.putAll(conf);
}
return this;
}
@Override
- public Map getRASConfiguration() {
- //TODO this should be read only
- return component.componentConf;
+ public BoltDeclarer addResources(Map<String, Double> resources) {
+ if (resources != null) {
+ Map<String, Double> currentResources = (Map<String, Double>) component.componentConf.computeIfAbsent(
+ Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, (k) -> new HashMap<>());
+ currentResources.putAll(resources);
+ }
+ return this;
}
-
@Override
public BoltDeclarer addSharedMemory(SharedMemory request) {
component.sharedMemory.add(request);
@@ -573,11 +576,10 @@ public class TransactionalTopologyBuilder {
@SuppressWarnings("unchecked")
@Override
public BoltDeclarer addResource(String resourceName, Number resourceValue) {
- Map<String, Double> resourcesMap = (Map<String, Double>) getRASConfiguration().get(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP);
+ Map<String, Double> resourcesMap = (Map<String, Double>) component.componentConf.computeIfAbsent(
+ Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, (k) -> new HashMap<>());
resourcesMap.put(resourceName, resourceValue.doubleValue());
-
- getRASConfiguration().put(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, resourcesMap);
return this;
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/dcfbf274/storm-client/src/jvm/org/apache/storm/trident/topology/TridentTopologyBuilder.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/trident/topology/TridentTopologyBuilder.java b/storm-client/src/jvm/org/apache/storm/trident/topology/TridentTopologyBuilder.java
index 274e913..39c52f9 100644
--- a/storm-client/src/jvm/org/apache/storm/trident/topology/TridentTopologyBuilder.java
+++ b/storm-client/src/jvm/org/apache/storm/trident/topology/TridentTopologyBuilder.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.trident.topology;
import org.apache.storm.Config;
@@ -51,14 +52,14 @@ import org.apache.storm.trident.topology.TridentBoltExecutor.CoordType;
// based on transactional topologies
public class TridentTopologyBuilder {
- Map<GlobalStreamId, String> batchIds = new HashMap();
- Map<String, TransactionalSpoutComponent> spouts = new HashMap();
- Map<String, SpoutComponent> _batchPerTupleSpouts = new HashMap();
- Map<String, Component> _bolts = new HashMap();
+ Map<GlobalStreamId, String> batchIds = new HashMap<>();
+ Map<String, TransactionalSpoutComponent> spouts = new HashMap<>();
+ Map<String, SpoutComponent> _batchPerTupleSpouts = new HashMap<>();
+ Map<String, Component> bolts = new HashMap<>();
public SpoutDeclarer setBatchPerTupleSpout(String id, String streamName, IRichSpout spout, Integer parallelism, String batchGroup) {
- Map<String, String> batchGroups = new HashMap();
+ Map<String, String> batchGroups = new HashMap<>();
batchGroups.put(streamName, batchGroup);
markBatchGroups(id, batchGroups);
SpoutComponent c = new SpoutComponent(spout, streamName, parallelism, batchGroup);
@@ -71,7 +72,7 @@ public class TridentTopologyBuilder {
}
public SpoutDeclarer setSpout(String id, String streamName, String txStateId, ITridentSpout spout, Integer parallelism, String batchGroup) {
- Map<String, String> batchGroups = new HashMap();
+ Map<String, String> batchGroups = new HashMap<>();
batchGroups.put(streamName, batchGroup);
markBatchGroups(id, batchGroups);
@@ -84,7 +85,7 @@ public class TridentTopologyBuilder {
public BoltDeclarer setBolt(String id, ITridentBatchBolt bolt, Integer parallelism, Set<String> committerBatches, Map<String, String> batchGroups) {
markBatchGroups(id, batchGroups);
Component c = new Component(bolt, parallelism, committerBatches);
- _bolts.put(id, c);
+ bolts.put(id, c);
return new BoltDeclarerImpl(c);
}
@@ -105,7 +106,7 @@ public class TridentTopologyBuilder {
Map<GlobalStreamId, String> fleshOutStreamBatchIds(boolean includeCommitStream) {
Map<GlobalStreamId, String> ret = new HashMap<>(batchIds);
- Set<String> allBatches = new HashSet(batchIds.values());
+ Set<String> allBatches = new HashSet<>(batchIds.values());
for(String b: allBatches) {
ret.put(new GlobalStreamId(masterCoordinator(b), MasterBatchCoordinator.BATCH_STREAM_ID), b);
if(includeCommitStream) {
@@ -165,11 +166,9 @@ public class TridentTopologyBuilder {
for (SharedMemory request: c.sharedMemory) {
scd.addSharedMemory(request);
}
- for(Map<String, Object> m: c.componentConfs) {
- scd.addConfigurations(m);
- }
+ scd.addConfigurations(c.componentConf);
- Map<String, TridentBoltExecutor.CoordSpec> specs = new HashMap();
+ Map<String, TridentBoltExecutor.CoordSpec> specs = new HashMap<>();
specs.put(c.batchGroupId, new CoordSpec());
BoltDeclarer bd = builder.setBolt(id,
new TridentBoltExecutor(
@@ -182,22 +181,18 @@ public class TridentTopologyBuilder {
c.parallelism);
bd.allGrouping(spoutCoordinator(id), MasterBatchCoordinator.BATCH_STREAM_ID);
bd.allGrouping(masterCoordinator(batchGroup), MasterBatchCoordinator.SUCCESS_STREAM_ID);
- if(c.spout instanceof ICommitterTridentSpout) {
+ if (c.spout instanceof ICommitterTridentSpout) {
bd.allGrouping(masterCoordinator(batchGroup), MasterBatchCoordinator.COMMIT_STREAM_ID);
}
- for(Map<String, Object> m: c.componentConfs) {
- bd.addConfigurations(m);
- }
+ bd.addConfigurations(c.componentConf);
}
}
for(String id: _batchPerTupleSpouts.keySet()) {
SpoutComponent c = _batchPerTupleSpouts.get(id);
SpoutDeclarer d = builder.setSpout(id, new RichSpoutBatchTriggerer((IRichSpout) c.spout, c.streamName, c.batchGroupId), c.parallelism);
-
- for(Map<String, Object> conf: c.componentConfs) {
- d.addConfigurations(conf);
- }
+
+ d.addConfigurations(c.componentConf);
}
Number onHeap = masterCoordResources.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB);
@@ -222,8 +217,8 @@ public class TridentTopologyBuilder {
}
}
- for(String id: _bolts.keySet()) {
- Component c = _bolts.get(id);
+ for(String id: bolts.keySet()) {
+ Component c = bolts.get(id);
Map<String, CoordSpec> specs = new HashMap<>();
@@ -248,9 +243,7 @@ public class TridentTopologyBuilder {
for (SharedMemory request: c.sharedMemory) {
d.addSharedMemory(request);
}
- for(Map<String, Object> conf: c.componentConfs) {
- d.addConfigurations(conf);
- }
+ d.addConfigurations(c.componentConf);
for(InputDeclaration inputDecl: c.declarations) {
inputDecl.declare(d);
@@ -281,7 +274,7 @@ public class TridentTopologyBuilder {
private static class SpoutComponent {
public final Object spout;
public final Integer parallelism;
- public final List<Map<String, Object>> componentConfs = new ArrayList<>();
+ public final Map<String, Object> componentConf = new HashMap<>();
final String batchGroupId;
final String streamName;
final Set<SharedMemory> sharedMemory = new HashSet<>();
@@ -317,7 +310,7 @@ public class TridentTopologyBuilder {
public final ITridentBatchBolt bolt;
public final Integer parallelism;
public final List<InputDeclaration> declarations = new ArrayList<>();
- public final List<Map<String, Object>> componentConfs = new ArrayList<>();
+ public final Map<String, Object> componentConf = new HashMap<>();
public final Set<String> committerBatches;
public final Set<SharedMemory> sharedMemory = new HashSet<>();
@@ -334,18 +327,18 @@ public class TridentTopologyBuilder {
}
Map<String, Set<String>> getBoltBatchToComponentSubscriptions(String id) {
- Map<String, Set<String>> ret = new HashMap();
+ Map<String, Set<String>> ret = new HashMap<>();
for(GlobalStreamId s: getBoltSubscriptionStreams(id)) {
String b = batchIds.get(s);
- if(!ret.containsKey(b)) ret.put(b, new HashSet());
+ if(!ret.containsKey(b)) ret.put(b, new HashSet<>());
ret.get(b).add(s.get_componentId());
}
return ret;
}
List<GlobalStreamId> getBoltSubscriptionStreams(String id) {
- List<GlobalStreamId> ret = new ArrayList();
- Component c = _bolts.get(id);
+ List<GlobalStreamId> ret = new ArrayList<>();
+ Component c = bolts.get(id);
for(InputDeclaration d: c.declarations) {
ret.add(new GlobalStreamId(d.getComponent(), d.getStream()));
}
@@ -359,54 +352,52 @@ public class TridentTopologyBuilder {
}
private static class SpoutDeclarerImpl extends BaseConfigurationDeclarer<SpoutDeclarer> implements SpoutDeclarer {
- SpoutComponent _component;
+ SpoutComponent component;
public SpoutDeclarerImpl(SpoutComponent component) {
- _component = component;
+ this.component = component;
}
@Override
public SpoutDeclarer addConfigurations(Map<String, Object> conf) {
- _component.componentConfs.add(conf);
+ if (conf != null) {
+ component.componentConf.putAll(conf);
+ }
+ return this;
+ }
+
+ @Override
+ public SpoutDeclarer addResources(Map<String, Double> resources) {
+ if (resources != null) {
+ Map<String, Double> currentResources = (Map<String, Double>) component.componentConf.computeIfAbsent(
+ Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, (k) -> new HashMap<>());
+ currentResources.putAll(resources);
+ }
return this;
}
@SuppressWarnings("unchecked")
@Override
public SpoutDeclarer addResource(String resourceName, Number resourceValue) {
- Map<String, Double> resourcesMap = (Map<String, Double>) getRASConfiguration().get(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP);
+ Map<String, Double> resourcesMap = (Map<String, Double>) component.componentConf.computeIfAbsent(
+ Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, (k) -> new HashMap<>());
resourcesMap.put(resourceName, resourceValue.doubleValue());
-
- getRASConfiguration().put(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, resourcesMap);
return this;
}
@Override
- public Map getRASConfiguration() {
- for(Map<String, Object> conf : _component.componentConfs) {
- if (conf.containsKey(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP)) {
- return conf;
- }
- }
- Map<String, Object> newConf = new HashMap<>();
- newConf.put(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, new HashMap());
- _component.componentConfs.add(newConf);
- return newConf;
- }
-
- @Override
public SpoutDeclarer addSharedMemory(SharedMemory request) {
- _component.sharedMemory.add(request);
+ component.sharedMemory.add(request);
return this;
}
}
private static class BoltDeclarerImpl extends BaseConfigurationDeclarer<BoltDeclarer> implements BoltDeclarer {
- Component _component;
+ Component component;
public BoltDeclarerImpl(Component component) {
- _component = component;
+ this.component = component;
}
@Override
@@ -777,42 +768,40 @@ public class TridentTopologyBuilder {
}
private void addDeclaration(InputDeclaration declaration) {
- _component.declarations.add(declaration);
+ component.declarations.add(declaration);
}
@Override
public BoltDeclarer addConfigurations(Map<String, Object> conf) {
- _component.componentConfs.add(conf);
+ if (conf != null) {
+ component.componentConf.putAll(conf);
+ }
return this;
}
@Override
- public Map getRASConfiguration() {
- for(Map<String, Object> conf : _component.componentConfs) {
- if (conf.containsKey(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP)) {
- return conf;
- }
+ public BoltDeclarer addResources(Map<String, Double> resources) {
+ if (resources != null) {
+ Map<String, Double> currentResources = (Map<String, Double>) component.componentConf.computeIfAbsent(
+ Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, (k) -> new HashMap<>());
+ currentResources.putAll(resources);
}
- Map<String, Object> newConf = new HashMap<>();
- newConf.put(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, new HashMap());
- _component.componentConfs.add(newConf);
- return newConf;
+ return this;
}
@Override
public BoltDeclarer addSharedMemory(SharedMemory request) {
- _component.sharedMemory.add(request);
+ component.sharedMemory.add(request);
return this;
}
@SuppressWarnings("unchecked")
@Override
public BoltDeclarer addResource(String resourceName, Number resourceValue) {
- Map<String, Double> resourcesMap = (Map<String, Double>) getRASConfiguration().get(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP);
+ Map<String, Double> resourcesMap = (Map<String, Double>) component.componentConf.computeIfAbsent(
+ Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, (k) -> new HashMap<>());
resourcesMap.put(resourceName, resourceValue.doubleValue());
-
- getRASConfiguration().put(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, resourcesMap);
return this;
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/dcfbf274/storm-server/pom.xml
----------------------------------------------------------------------
diff --git a/storm-server/pom.xml b/storm-server/pom.xml
index 9a1b4c0..18fec7f 100644
--- a/storm-server/pom.xml
+++ b/storm-server/pom.xml
@@ -130,7 +130,7 @@
<artifactId>maven-checkstyle-plugin</artifactId>
<!--Note - the version would be inherited-->
<configuration>
- <maxAllowedViolations>2699</maxAllowedViolations>
+ <maxAllowedViolations>2655</maxAllowedViolations>
</configuration>
</plugin>
<plugin>
http://git-wip-us.apache.org/repos/asf/storm/blob/dcfbf274/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java b/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java
index 69c8756..ba1c618 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java
@@ -500,8 +500,7 @@ public class TopologyDetails {
Map<String,Double> topologyComponentResourcesMap = (
Map<String, Double>) this.topologyConf.getOrDefault(
- Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, new HashMap<>()
- );
+ Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, new HashMap<>());
topologyComponentResourcesMap = normalizedResourceMap(topologyComponentResourcesMap);
http://git-wip-us.apache.org/repos/asf/storm/blob/dcfbf274/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceUtils.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceUtils.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceUtils.java
index c13dce6..9b0ee15 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceUtils.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceUtils.java
@@ -26,12 +26,10 @@ import java.util.Map;
import java.util.Set;
import org.apache.storm.Config;
-import org.apache.storm.Constants;
import org.apache.storm.generated.Bolt;
import org.apache.storm.generated.ComponentCommon;
import org.apache.storm.generated.SpoutSpec;
import org.apache.storm.generated.StormTopology;
-import org.apache.storm.utils.ConfigUtils;
import org.apache.storm.utils.ObjectReader;
import org.json.simple.JSONObject;
import org.json.simple.parser.JSONParser;
@@ -159,7 +157,7 @@ public class ResourceUtils {
Map<String, Double> topologyComponentResourcesMap =
(Map<String, Double>) topologyConf.getOrDefault(
- Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, Collections.emptyMap());
+ Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, new HashMap<>());
resourceNameSet.addAll(topologyResources.keySet());
resourceNameSet.addAll(topologyComponentResourcesMap.keySet());
@@ -232,8 +230,8 @@ public class ResourceUtils {
// If resource is also present in resources map will overwrite the above
if (jsonObject.containsKey(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP)) {
Map<String, Number> rawResourcesMap =
- (Map<String, Number>) jsonObject.getOrDefault(
- Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, Collections.emptyMap());
+ (Map<String, Number>) jsonObject.computeIfAbsent(
+ Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, (k) -> new HashMap<>());
for (Map.Entry<String, Number> stringNumberEntry : rawResourcesMap.entrySet()) {
topologyResources.put(
http://git-wip-us.apache.org/repos/asf/storm/blob/dcfbf274/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java
----------------------------------------------------------------------
diff --git a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java
index b8d8206..60c314f 100644
--- a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java
+++ b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java
@@ -18,6 +18,7 @@
package org.apache.storm.scheduler.resource.strategies.scheduling;
+import java.util.Set;
import org.apache.storm.Config;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.generated.WorkerResources;
@@ -67,7 +68,7 @@ public class TestGenericResourceAwareStrategy {
private static int currentTime = 1450418597;
/**
- * test if the scheduling logic for the GenericResourceAwareStrategy is correct
+ * test if the scheduling logic for the GenericResourceAwareStrategy is correct.
*/
@Test
public void testGenericResourceAwareStrategySharedMemory() {
@@ -95,7 +96,7 @@ public class TestGenericResourceAwareStrategy {
INimbus iNimbus = new INimbusTest();
Config conf = createGrasClusterConfig(cpuPercent, memoryOnHeap, memoryOffHeap, null, Collections.emptyMap());
- Map<String, Double> genericResourcesMap = new HashMap();
+ Map<String, Double> genericResourcesMap = new HashMap<>();
genericResourcesMap.put("gpu.count", 1.0);
Map<String, SupervisorDetails> supMap = genSupervisors(4, 4, 500, 2000, genericResourcesMap);
@@ -129,23 +130,34 @@ public class TestGenericResourceAwareStrategy {
double totalExpectedWorkerOffHeap = (totalNumberOfTasks * memoryOffHeap) + sharedOffHeapWorker;
SchedulerAssignment assignment = cluster.getAssignmentById(topo.getId());
- assertEquals(1, assignment.getSlots().size());
- WorkerSlot ws = assignment.getSlots().iterator().next();
- String nodeId = ws.getNodeId();
- assertEquals(1, assignment.getNodeIdToTotalSharedOffHeapMemory().size());
- assertEquals(sharedOffHeapNode, assignment.getNodeIdToTotalSharedOffHeapMemory().get(nodeId), 0.01);
- assertEquals(1, assignment.getScheduledResources().size());
- WorkerResources resources = assignment.getScheduledResources().get(ws);
- assertEquals(totalExpectedCPU, resources.get_cpu(), 0.01);
- assertEquals(totalExpectedOnHeap, resources.get_mem_on_heap(), 0.01);
- assertEquals(totalExpectedWorkerOffHeap, resources.get_mem_off_heap(), 0.01);
- assertEquals(sharedOnHeap, resources.get_shared_mem_on_heap(), 0.01);
- assertEquals(sharedOffHeapWorker, resources.get_shared_mem_off_heap(), 0.01);
+ Set<WorkerSlot> slots = assignment.getSlots();
+ Map<String, Double> nodeToTotalShared = assignment.getNodeIdToTotalSharedOffHeapMemory();
+ LOG.info("NODE TO SHARED OFF HEAP {}", nodeToTotalShared);
+ Map<WorkerSlot, WorkerResources> scheduledResources = assignment.getScheduledResources();
+ assertEquals(2, slots.size());
+ assertEquals(2, nodeToTotalShared.size());
+ assertEquals(2, scheduledResources.size());
+ double totalFoundCPU = 0.0;
+ double totalFoundOnHeap = 0.0;
+ double totalFoundWorkerOffHeap = 0.0;
+ for (WorkerSlot ws : slots) {
+ WorkerResources resources = scheduledResources.get(ws);
+ totalFoundCPU += resources.get_cpu();
+ totalFoundOnHeap += resources.get_mem_on_heap();
+ totalFoundWorkerOffHeap += resources.get_mem_off_heap();
+ }
+
+ assertEquals(totalExpectedCPU, totalFoundCPU, 0.01);
+ assertEquals(totalExpectedOnHeap, totalFoundOnHeap, 0.01);
+ assertEquals(totalExpectedWorkerOffHeap, totalFoundWorkerOffHeap, 0.01);
+ assertEquals(sharedOffHeapNode, nodeToTotalShared.values().stream().mapToDouble((d) -> d).sum(), 0.01);
+ assertEquals(sharedOnHeap, scheduledResources.values().stream().mapToDouble((r) -> r.get_shared_mem_on_heap()).sum(), 0.01);
+ assertEquals(sharedOffHeapWorker, scheduledResources.values().stream().mapToDouble((r) -> r.get_shared_mem_off_heap()).sum(),
+ 0.01);
}
-
-
+
/**
- * test if the scheduling logic for the GenericResourceAwareStrategy is correct
+ * test if the scheduling logic for the GenericResourceAwareStrategy is correct.
*/
@Test
public void testGenericResourceAwareStrategy() {
@@ -187,16 +199,21 @@ public class TestGenericResourceAwareStrategy {
rs.prepare(conf);
rs.schedule(topologies, cluster);
+ //We need to have 3 slots on 3 separate hosts to make the GPU situation work
+
HashSet<HashSet<ExecutorDetails>> expectedScheduling = new HashSet<>();
- expectedScheduling.add(new HashSet<>(Arrays.asList(new ExecutorDetails(0, 0)))); //Spout
+ expectedScheduling.add(new HashSet<>(Arrays.asList(new ExecutorDetails(3, 3)))); //bolt-3 - 500 MB, 50% CPU, 2 GPU
+ //Total 500 MB, 50% CPU, 2 - GPU -> this node has 1000 MB, 100% cpu, 0 GPU left
expectedScheduling.add(new HashSet<>(Arrays.asList(
- new ExecutorDetails(2, 2), //bolt-1
- new ExecutorDetails(4, 4), //bolt-2
- new ExecutorDetails(6, 6)))); //bolt-3
+ new ExecutorDetails(2, 2), //bolt-1 - 500 MB, 50% CPU, 0 GPU
+ new ExecutorDetails(5, 5), //bolt-2 - 500 MB, 50% CPU, 1 GPU
+ new ExecutorDetails(6, 6)))); //bolt-2 - 500 MB, 50% CPU, 1 GPU
+ //Total 1500 MB, 150% CPU, 2 GPU -> this node has 0 MB, 0% CPU, 0 GPU left
expectedScheduling.add(new HashSet<>(Arrays.asList(
- new ExecutorDetails(1, 1), //bolt-1
- new ExecutorDetails(3, 3), //bolt-2
- new ExecutorDetails(5, 5)))); //bolt-3
+ new ExecutorDetails(0, 0), //Spout - 500 MB, 50% CPU, 0 GPU
+ new ExecutorDetails(1, 1), //bolt-1 - 500 MB, 50% CPU, 0 GPU
+ new ExecutorDetails(4, 4)))); //bolt-3 500 MB, 50% cpu, 2 GPU
+ //Total 1500 MB, 150% CPU, 2 GPU -> this node has 0 MB, 0% CPU, 0 GPU left
HashSet<HashSet<ExecutorDetails>> foundScheduling = new HashSet<>();
SchedulerAssignment assignment = cluster.getAssignmentById("testTopology-id");
for (Collection<ExecutorDetails> execs : assignment.getSlotToExecutors().values()) {
@@ -205,5 +222,4 @@ public class TestGenericResourceAwareStrategy {
Assert.assertEquals(expectedScheduling, foundScheduling);
}
-
}
[3/4] storm git commit: Addressed review comments
Posted by bo...@apache.org.
Addressed review comments
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/14cf5e61
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/14cf5e61
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/14cf5e61
Branch: refs/heads/master
Commit: 14cf5e61a96775d1997ec5e79d22fb3a36f13b6b
Parents: dcfbf27
Author: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Authored: Fri Nov 17 09:34:15 2017 -0600
Committer: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Committed: Fri Nov 17 09:34:15 2017 -0600
----------------------------------------------------------------------
.../apache/storm/topology/TopologyBuilder.java | 2 +-
.../topology/MasterBatchCoordinator.java | 2 +-
.../apache/storm/scheduler/TopologyDetails.java | 4 +-
.../TestGenericResourceAwareStrategy.java | 52 ++++++++++----------
4 files changed, 29 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/14cf5e61/storm-client/src/jvm/org/apache/storm/topology/TopologyBuilder.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/topology/TopologyBuilder.java b/storm-client/src/jvm/org/apache/storm/topology/TopologyBuilder.java
index f2e09b8..7c28797 100644
--- a/storm-client/src/jvm/org/apache/storm/topology/TopologyBuilder.java
+++ b/storm-client/src/jvm/org/apache/storm/topology/TopologyBuilder.java
@@ -593,7 +593,7 @@ public class TopologyBuilder {
Map<String, Double> currentResources =
(Map<String, Double>) conf.computeIfAbsent(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, (k) -> new HashMap<>());
currentResources.putAll(resources);
- commons.get(id).set_json_conf(JSONValue.toJSONString(currConf));
+ commons.get(id).set_json_conf(JSONValue.toJSONString(conf));
}
return (T) this;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/14cf5e61/storm-client/src/jvm/org/apache/storm/trident/topology/MasterBatchCoordinator.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/trident/topology/MasterBatchCoordinator.java b/storm-client/src/jvm/org/apache/storm/trident/topology/MasterBatchCoordinator.java
index 3417112..e54d0e2 100644
--- a/storm-client/src/jvm/org/apache/storm/trident/topology/MasterBatchCoordinator.java
+++ b/storm-client/src/jvm/org/apache/storm/trident/topology/MasterBatchCoordinator.java
@@ -308,7 +308,7 @@ public class MasterBatchCoordinator extends BaseRichSpout {
", _maxTransactionActive=" + _maxTransactionActive +
", _coordinators=" + _coordinators +
", _managedSpoutIds=" + _managedSpoutIds +
- ", spouts=" + _spouts +
+ ", _spouts=" + _spouts +
", _throttler=" + _throttler +
", _active=" + _active +
"}";
http://git-wip-us.apache.org/repos/asf/storm/blob/14cf5e61/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java b/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java
index ba1c618..d92c80a 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/TopologyDetails.java
@@ -25,7 +25,6 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-
import org.apache.storm.Config;
import org.apache.storm.Constants;
import org.apache.storm.daemon.Acker;
@@ -37,7 +36,6 @@ import org.apache.storm.generated.SharedMemory;
import org.apache.storm.generated.SpoutSpec;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.scheduler.resource.ResourceUtils;
-import org.apache.storm.utils.ConfigUtils;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
import org.apache.storm.utils.Utils;
@@ -334,7 +332,7 @@ public class TopologyDetails {
/**
* Get the total resource requirement for an executor.
- * @param exec
+ * @param exec the executor to get the resources for.
* @return Double the total about of cpu requirement for executor
*/
public Map<String, Double> getTotalResources(ExecutorDetails exec) {
http://git-wip-us.apache.org/repos/asf/storm/blob/14cf5e61/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java
----------------------------------------------------------------------
diff --git a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java
index 60c314f..fc44568 100644
--- a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java
+++ b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -6,9 +6,9 @@
* 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
- * <p>
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
* 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.
@@ -18,11 +18,17 @@
package org.apache.storm.scheduler.resource.strategies.scheduling;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Map.Entry;
import java.util.Set;
import org.apache.storm.Config;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.generated.WorkerResources;
-import org.apache.storm.networktopography.DNSToSwitchMapping;
import org.apache.storm.scheduler.Cluster;
import org.apache.storm.scheduler.ExecutorDetails;
import org.apache.storm.scheduler.INimbus;
@@ -33,10 +39,7 @@ import org.apache.storm.scheduler.SupervisorResources;
import org.apache.storm.scheduler.Topologies;
import org.apache.storm.scheduler.TopologyDetails;
import org.apache.storm.scheduler.WorkerSlot;
-import org.apache.storm.scheduler.resource.RAS_Node;
import org.apache.storm.scheduler.resource.ResourceAwareScheduler;
-import org.apache.storm.scheduler.resource.SchedulingResult;
-import org.apache.storm.scheduler.resource.strategies.scheduling.BaseResourceAwareStrategy.ObjectResources;
import org.apache.storm.topology.SharedOffHeapWithinNode;
import org.apache.storm.topology.SharedOffHeapWithinWorker;
import org.apache.storm.topology.SharedOnHeap;
@@ -46,21 +49,9 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.junit.Assert.*;
import static org.apache.storm.scheduler.resource.TestUtilsForResourceAwareScheduler.*;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.TreeSet;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
public class TestGenericResourceAwareStrategy {
private static final Logger LOG = LoggerFactory.getLogger(TestGenericResourceAwareStrategy.class);
@@ -123,7 +114,12 @@ public class TestGenericResourceAwareStrategy {
assertTrue(supervisorId, resources.getTotalMem() >= resources.getUsedMem());
}
- // Everything should fit in a single slot
+ // If we didn't take GPUs into account everything would fit under a single slot
+ // But because there is only 1 GPU per node, and each of the 2 spouts needs a GPU
+ // It has to be scheduled on at least 2 nodes, and hence 2 slots.
+ // Because of this all of the bolts will be scheduled on a single slot with one of
+ // the spouts and the other spout is on its own slot. So everything that can be shared is
+ // shared.
int totalNumberOfTasks = (spoutParallelism + (boltParallelism * numBolts));
double totalExpectedCPU = totalNumberOfTasks * cpuPercent;
double totalExpectedOnHeap = (totalNumberOfTasks * memoryOnHeap) + sharedOnHeap;
@@ -151,8 +147,8 @@ public class TestGenericResourceAwareStrategy {
assertEquals(totalExpectedOnHeap, totalFoundOnHeap, 0.01);
assertEquals(totalExpectedWorkerOffHeap, totalFoundWorkerOffHeap, 0.01);
assertEquals(sharedOffHeapNode, nodeToTotalShared.values().stream().mapToDouble((d) -> d).sum(), 0.01);
- assertEquals(sharedOnHeap, scheduledResources.values().stream().mapToDouble((r) -> r.get_shared_mem_on_heap()).sum(), 0.01);
- assertEquals(sharedOffHeapWorker, scheduledResources.values().stream().mapToDouble((r) -> r.get_shared_mem_off_heap()).sum(),
+ assertEquals(sharedOnHeap, scheduledResources.values().stream().mapToDouble(WorkerResources::get_shared_mem_on_heap).sum(), 0.01);
+ assertEquals(sharedOffHeapWorker, scheduledResources.values().stream().mapToDouble(WorkerResources::get_shared_mem_off_heap).sum(),
0.01);
}
@@ -178,7 +174,7 @@ public class TestGenericResourceAwareStrategy {
INimbus iNimbus = new INimbusTest();
Config conf = createGrasClusterConfig(50, 250, 250, null, Collections.emptyMap());
- Map<String, Double> genericResourcesMap = new HashMap();
+ Map<String, Double> genericResourcesMap = new HashMap<>();
genericResourcesMap.put("gpu.count", 2.0);
Map<String, SupervisorDetails> supMap = genSupervisors(4, 4, 150, 1500, genericResourcesMap);
@@ -199,7 +195,11 @@ public class TestGenericResourceAwareStrategy {
rs.prepare(conf);
rs.schedule(topologies, cluster);
- //We need to have 3 slots on 3 separate hosts to make the GPU situation work
+ //We need to have 3 slots on 3 separate hosts. The topology needs 6 GPUs 3500 MB memory and 350% CPU
+ // The bolt-3 instances must be on separate nodes because they each need 2 GPUs.
+ // The bolt-2 instances must be on the same node as they each need 1 GPU
+ // (this assumes that we are packing the components to avoid fragmentation).
+ // The bolt-1 and spout instances fill in the rest.
HashSet<HashSet<ExecutorDetails>> expectedScheduling = new HashSet<>();
expectedScheduling.add(new HashSet<>(Arrays.asList(new ExecutorDetails(3, 3)))); //bolt-3 - 500 MB, 50% CPU, 2 GPU
[2/4] storm git commit: STORM-2805: Clean up confs in TopologyBuilders
Posted by bo...@apache.org.
STORM-2805: Clean up confs in TopologyBuilders
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/96c461d0
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/96c461d0
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/96c461d0
Branch: refs/heads/master
Commit: 96c461d06274fbeaaccbe713967bdfd0585e3a61
Parents: 8e4a146
Author: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Authored: Mon Nov 13 07:58:06 2017 -0600
Committer: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Committed: Fri Nov 17 07:32:23 2017 -0600
----------------------------------------------------------------------
storm-client/pom.xml | 2 +-
.../coordination/BatchSubtopologyBuilder.java | 63 ++++++-------
.../storm/drpc/LinearDRPCTopologyBuilder.java | 48 +++++-----
.../TransactionalTopologyBuilder.java | 96 +++++++++-----------
.../topology/MasterBatchCoordinator.java | 2 +-
.../topology/TridentTopologyBuilder.java | 26 +++---
6 files changed, 108 insertions(+), 129 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/96c461d0/storm-client/pom.xml
----------------------------------------------------------------------
diff --git a/storm-client/pom.xml b/storm-client/pom.xml
index 19354c1..8a7efc4 100644
--- a/storm-client/pom.xml
+++ b/storm-client/pom.xml
@@ -261,7 +261,7 @@
<!--Note - the version would be inherited-->
<configuration>
<excludes>**/generated/**</excludes>
- <maxAllowedViolations>10473</maxAllowedViolations>
+ <maxAllowedViolations>10320</maxAllowedViolations>
</configuration>
</plugin>
<plugin>
http://git-wip-us.apache.org/repos/asf/storm/blob/96c461d0/storm-client/src/jvm/org/apache/storm/coordination/BatchSubtopologyBuilder.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/coordination/BatchSubtopologyBuilder.java b/storm-client/src/jvm/org/apache/storm/coordination/BatchSubtopologyBuilder.java
index 9318aa8..9b1495d 100644
--- a/storm-client/src/jvm/org/apache/storm/coordination/BatchSubtopologyBuilder.java
+++ b/storm-client/src/jvm/org/apache/storm/coordination/BatchSubtopologyBuilder.java
@@ -44,14 +44,14 @@ import org.apache.storm.tuple.Fields;
public class BatchSubtopologyBuilder {
- Map<String, Component> _bolts = new HashMap<String, Component>();
- Component _masterBolt;
- String _masterId;
+ Map<String, Component> bolts = new HashMap<>();
+ Component masterBolt;
+ String masterId;
public BatchSubtopologyBuilder(String masterBoltId, IBasicBolt masterBolt, Number boltParallelism) {
Integer p = boltParallelism == null ? null : boltParallelism.intValue();
- _masterBolt = new Component(new BasicBoltExecutor(masterBolt), p);
- _masterId = masterBoltId;
+ this.masterBolt = new Component(new BasicBoltExecutor(masterBolt), p);
+ masterId = masterBoltId;
}
public BatchSubtopologyBuilder(String masterBoltId, IBasicBolt masterBolt) {
@@ -59,7 +59,7 @@ public class BatchSubtopologyBuilder {
}
public BoltDeclarer getMasterDeclarer() {
- return new BoltDeclarerImpl(_masterBolt);
+ return new BoltDeclarerImpl(masterBolt);
}
public BoltDeclarer setBolt(String id, IBatchBolt bolt) {
@@ -84,24 +84,24 @@ public class BatchSubtopologyBuilder {
p = parallelism.intValue();
}
Component component = new Component(bolt, p);
- _bolts.put(id, component);
+ bolts.put(id, component);
return new BoltDeclarerImpl(component);
}
public void extendTopology(TopologyBuilder builder) {
- BoltDeclarer declarer = builder.setBolt(_masterId, new CoordinatedBolt(_masterBolt.bolt), _masterBolt.parallelism);
- for (InputDeclaration decl: _masterBolt.declarations) {
+ BoltDeclarer declarer = builder.setBolt(masterId, new CoordinatedBolt(masterBolt.bolt), masterBolt.parallelism);
+ for (InputDeclaration decl: masterBolt.declarations) {
decl.declare(declarer);
}
- for (Map<String, Object> conf: _masterBolt.componentConfs) {
- declarer.addConfigurations(conf);
+ if (!masterBolt.componentConf.isEmpty()) {
+ declarer.addConfigurations(masterBolt.componentConf);
}
- for (String id: _bolts.keySet()) {
- Component component = _bolts.get(id);
+ for (String id: bolts.keySet()) {
+ Component component = bolts.get(id);
Map<String, SourceArgs> coordinatedArgs = new HashMap<String, SourceArgs>();
for (String c: componentBoltSubscriptions(component)) {
SourceArgs source;
- if (c.equals(_masterId)) {
+ if (c.equals(masterId)) {
source = SourceArgs.single();
} else {
source = SourceArgs.all();
@@ -118,8 +118,8 @@ public class BatchSubtopologyBuilder {
for (SharedMemory request: component.sharedMemory) {
input.addSharedMemory(request);
}
- for (Map<String, Object> conf: component.componentConfs) {
- input.addConfigurations(conf);
+ if (!component.componentConf.isEmpty()) {
+ input.addConfigurations(component.componentConf);
}
for (String c: componentBoltSubscriptions(component)) {
input.directGrouping(c, Constants.COORDINATED_STREAM_ID);
@@ -131,7 +131,7 @@ public class BatchSubtopologyBuilder {
}
private Set<String> componentBoltSubscriptions(Component component) {
- Set<String> ret = new HashSet<String>();
+ Set<String> ret = new HashSet<>();
for (InputDeclaration d: component.declarations) {
ret.add(d.getComponent());
}
@@ -142,7 +142,7 @@ public class BatchSubtopologyBuilder {
public final IRichBolt bolt;
public final Integer parallelism;
public final List<InputDeclaration> declarations = new ArrayList<>();
- public final List<Map<String, Object>> componentConfs = new ArrayList<>();
+ public final Map<String, Object> componentConf = new HashMap<>();
public final Set<SharedMemory> sharedMemory = new HashSet<>();
public Component(IRichBolt bolt, Integer parallelism) {
@@ -151,16 +151,16 @@ public class BatchSubtopologyBuilder {
}
}
- private static interface InputDeclaration {
+ private interface InputDeclaration {
void declare(InputDeclarer declarer);
String getComponent();
}
private static class BoltDeclarerImpl extends BaseConfigurationDeclarer<BoltDeclarer> implements BoltDeclarer {
- Component _component;
+ Component component;
public BoltDeclarerImpl(Component component) {
- _component = component;
+ this.component = component;
}
@Override
@@ -446,31 +446,26 @@ public class BatchSubtopologyBuilder {
}
private void addDeclaration(InputDeclaration declaration) {
- _component.declarations.add(declaration);
+ component.declarations.add(declaration);
}
@Override
public BoltDeclarer addConfigurations(Map<String, Object> conf) {
- _component.componentConfs.add(conf);
+ if (conf != null && !conf.isEmpty()) {
+ component.componentConf.putAll(conf);
+ }
return this;
}
@Override
- public Map getRASConfiguration() {
- for (Map<String, Object> conf : _component.componentConfs) {
- if (conf.containsKey(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP)) {
- return conf;
- }
- }
- Map<String, Object> newConf = new HashMap<>();
- newConf.put(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, new HashMap());
- _component.componentConfs.add(newConf);
- return newConf;
+ public Map<String, Object> getRASConfiguration() {
+ //TODO this should not be modifiable
+ return component.componentConf;
}
@Override
public BoltDeclarer addSharedMemory(SharedMemory request) {
- _component.sharedMemory.add(request);
+ component.sharedMemory.add(request);
return this;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/96c461d0/storm-client/src/jvm/org/apache/storm/drpc/LinearDRPCTopologyBuilder.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/drpc/LinearDRPCTopologyBuilder.java b/storm-client/src/jvm/org/apache/storm/drpc/LinearDRPCTopologyBuilder.java
index 928a5fa..51db422 100644
--- a/storm-client/src/jvm/org/apache/storm/drpc/LinearDRPCTopologyBuilder.java
+++ b/storm-client/src/jvm/org/apache/storm/drpc/LinearDRPCTopologyBuilder.java
@@ -53,12 +53,12 @@ import org.apache.storm.tuple.Fields;
// Trident subsumes the functionality provided by this class, so it's deprecated
@Deprecated
public class LinearDRPCTopologyBuilder {
- String _function;
- List<Component> _components = new ArrayList<Component>();
+ String function;
+ List<Component> components = new ArrayList<>();
public LinearDRPCTopologyBuilder(String function) {
- _function = function;
+ this.function = function;
}
public LinearDRPCInputDeclarer addBolt(IBatchBolt bolt, Number parallelism) {
@@ -75,7 +75,7 @@ public class LinearDRPCTopologyBuilder {
parallelism = 1;
}
Component component = new Component(bolt, parallelism.intValue());
- _components.add(component);
+ components.add(component);
return new InputDeclarerImpl(component);
}
@@ -93,11 +93,11 @@ public class LinearDRPCTopologyBuilder {
}
public StormTopology createLocalTopology(ILocalDRPC drpc) {
- return createTopology(new DRPCSpout(_function, drpc));
+ return createTopology(new DRPCSpout(function, drpc));
}
public StormTopology createRemoteTopology() {
- return createTopology(new DRPCSpout(_function));
+ return createTopology(new DRPCSpout(function));
}
@@ -110,8 +110,8 @@ public class LinearDRPCTopologyBuilder {
builder.setBolt(PREPARE_ID, new PrepareRequest())
.noneGrouping(SPOUT_ID);
int i = 0;
- for (; i < _components.size();i++) {
- Component component = _components.get(i);
+ for (; i < components.size(); i++) {
+ Component component = components.get(i);
Map<String, SourceArgs> source = new HashMap<String, SourceArgs>();
if (i == 1) {
@@ -120,7 +120,7 @@ public class LinearDRPCTopologyBuilder {
source.put(boltId(i - 1), SourceArgs.all());
}
IdStreamSpec idSpec = null;
- if (i == _components.size() - 1 && component.bolt instanceof FinishedCallback) {
+ if (i == components.size() - 1 && component.bolt instanceof FinishedCallback) {
idSpec = IdStreamSpec.makeDetectSpec(PREPARE_ID, PrepareRequest.ID_STREAM);
}
BoltDeclarer declarer = builder.setBolt(
@@ -132,8 +132,8 @@ public class LinearDRPCTopologyBuilder {
declarer.addSharedMemory(request);
}
- for (Map<String, Object> conf: component.componentConfs) {
- declarer.addConfigurations(conf);
+ if (!component.componentConf.isEmpty()) {
+ declarer.addConfigurations(component.componentConf);
}
if (idSpec != null) {
@@ -157,7 +157,7 @@ public class LinearDRPCTopologyBuilder {
}
}
- IRichBolt lastBolt = _components.get(_components.size() - 1).bolt;
+ IRichBolt lastBolt = components.get(components.size() - 1).bolt;
OutputFieldsGetter getter = new OutputFieldsGetter();
lastBolt.declareOutputFields(getter);
Map<String, StreamInfo> streams = getter.getFieldsDeclaration();
@@ -188,7 +188,7 @@ public class LinearDRPCTopologyBuilder {
private static class Component {
public final IRichBolt bolt;
public final int parallelism;
- public final List<Map<String, Object>> componentConfs = new ArrayList<>();
+ public final Map<String, Object> componentConf = new HashMap<>();
public final List<InputDeclaration> declarations = new ArrayList<>();
public final Set<SharedMemory> sharedMemory = new HashSet<>();
@@ -203,10 +203,10 @@ public class LinearDRPCTopologyBuilder {
}
private static class InputDeclarerImpl extends BaseConfigurationDeclarer<LinearDRPCInputDeclarer> implements LinearDRPCInputDeclarer {
- Component _component;
+ Component component;
public InputDeclarerImpl(Component component) {
- _component = component;
+ this.component = component;
}
@Override
@@ -396,12 +396,14 @@ public class LinearDRPCTopologyBuilder {
}
private void addDeclaration(InputDeclaration declaration) {
- _component.declarations.add(declaration);
+ component.declarations.add(declaration);
}
@Override
public LinearDRPCInputDeclarer addConfigurations(Map<String, Object> conf) {
- _component.componentConfs.add(conf);
+ if (conf != null && !conf.isEmpty()) {
+ component.componentConf.putAll(conf);
+ }
return this;
}
@@ -418,20 +420,12 @@ public class LinearDRPCTopologyBuilder {
@Override
public Map getRASConfiguration() {
- for (Map<String, Object> conf : _component.componentConfs) {
- if (conf.containsKey(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP)) {
- return conf;
- }
- }
- Map<String, Object> newConf = new HashMap<>();
- newConf.put(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, new HashMap());
- _component.componentConfs.add(newConf);
- return newConf;
+ return component.componentConf;
}
@Override
public LinearDRPCInputDeclarer addSharedMemory(SharedMemory request) {
- _component.sharedMemory.add(request);
+ component.sharedMemory.add(request);
return this;
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/96c461d0/storm-client/src/jvm/org/apache/storm/transactional/TransactionalTopologyBuilder.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/transactional/TransactionalTopologyBuilder.java b/storm-client/src/jvm/org/apache/storm/transactional/TransactionalTopologyBuilder.java
index f2deca8..6f994ed 100644
--- a/storm-client/src/jvm/org/apache/storm/transactional/TransactionalTopologyBuilder.java
+++ b/storm-client/src/jvm/org/apache/storm/transactional/TransactionalTopologyBuilder.java
@@ -58,21 +58,21 @@ import java.util.Set;
*/
@Deprecated
public class TransactionalTopologyBuilder {
- final String _id;
- final String _spoutId;
- final ITransactionalSpout _spout;
- final Map<String, Component> _bolts = new HashMap<String, Component>();
- final Integer _spoutParallelism;
- final List<Map<String, Object>> _spoutConfs = new ArrayList<>();
- final Set<SharedMemory> _spoutSharedMemory = new HashSet<>();
+ final String id;
+ final String spoutId;
+ final ITransactionalSpout spout;
+ final Map<String, Component> bolts = new HashMap<>();
+ final Integer spoutParallelism;
+ final Map<String, Object> spoutConf = new HashMap<>();
+ final Set<SharedMemory> spoutSharedMemory = new HashSet<>();
// id is used to store the state of this transactionalspout in zookeeper
// it would be very dangerous to have 2 topologies active with the same id in the same cluster
public TransactionalTopologyBuilder(String id, String spoutId, ITransactionalSpout spout, Number spoutParallelism) {
- _id = id;
- _spoutId = spoutId;
- _spout = spout;
- _spoutParallelism = (spoutParallelism == null) ? null : spoutParallelism.intValue();
+ this.id = id;
+ this.spoutId = spoutId;
+ this.spout = spout;
+ this.spoutParallelism = (spoutParallelism == null) ? null : spoutParallelism.intValue();
}
public TransactionalTopologyBuilder(String id, String spoutId, ITransactionalSpout spout) {
@@ -127,35 +127,35 @@ public class TransactionalTopologyBuilder {
Integer p = null;
if(parallelism!=null) p = parallelism.intValue();
Component component = new Component(bolt, p, committer);
- _bolts.put(id, component);
+ bolts.put(id, component);
return new BoltDeclarerImpl(component);
}
public TopologyBuilder buildTopologyBuilder() {
- String coordinator = _spoutId + "/coordinator";
+ String coordinator = spoutId + "/coordinator";
TopologyBuilder builder = new TopologyBuilder();
- SpoutDeclarer declarer = builder.setSpout(coordinator, new TransactionalSpoutCoordinator(_spout));
- for (SharedMemory request: _spoutSharedMemory) {
+ SpoutDeclarer declarer = builder.setSpout(coordinator, new TransactionalSpoutCoordinator(spout));
+ for (SharedMemory request: spoutSharedMemory) {
declarer.addSharedMemory(request);
}
- for(Map<String, Object> conf: _spoutConfs) {
- declarer.addConfigurations(conf);
+ if (!spoutConf.isEmpty()) {
+ declarer.addConfigurations(spoutConf);
}
- declarer.addConfiguration(Config.TOPOLOGY_TRANSACTIONAL_ID, _id);
+ declarer.addConfiguration(Config.TOPOLOGY_TRANSACTIONAL_ID, id);
BoltDeclarer emitterDeclarer =
- builder.setBolt(_spoutId,
- new CoordinatedBolt(new TransactionalSpoutBatchExecutor(_spout),
+ builder.setBolt(spoutId,
+ new CoordinatedBolt(new TransactionalSpoutBatchExecutor(spout),
null,
null),
- _spoutParallelism)
+ spoutParallelism)
.allGrouping(coordinator, TransactionalSpoutCoordinator.TRANSACTION_BATCH_STREAM_ID)
- .addConfiguration(Config.TOPOLOGY_TRANSACTIONAL_ID, _id);
- if(_spout instanceof ICommitterTransactionalSpout) {
+ .addConfiguration(Config.TOPOLOGY_TRANSACTIONAL_ID, id);
+ if(spout instanceof ICommitterTransactionalSpout) {
emitterDeclarer.allGrouping(coordinator, TransactionalSpoutCoordinator.TRANSACTION_COMMIT_STREAM_ID);
}
- for(String id: _bolts.keySet()) {
- Component component = _bolts.get(id);
+ for(String id: bolts.keySet()) {
+ Component component = bolts.get(id);
Map<String, SourceArgs> coordinatedArgs = new HashMap<String, SourceArgs>();
for(String c: componentBoltSubscriptions(component)) {
coordinatedArgs.put(c, SourceArgs.all());
@@ -173,8 +173,8 @@ public class TransactionalTopologyBuilder {
for (SharedMemory request: component.sharedMemory) {
input.addSharedMemory(request);
}
- for(Map<String, Object> conf: component.componentConfs) {
- input.addConfigurations(conf);
+ if (!component.componentConf.isEmpty()) {
+ input.addConfigurations(component.componentConf);
}
for(String c: componentBoltSubscriptions(component)) {
input.directGrouping(c, Constants.COORDINATED_STREAM_ID);
@@ -205,7 +205,7 @@ public class TransactionalTopologyBuilder {
public final IRichBolt bolt;
public final Integer parallelism;
public final List<InputDeclaration> declarations = new ArrayList<>();
- public final List<Map<String, Object>> componentConfs = new ArrayList<>();
+ public final Map<String, Object> componentConf = new HashMap<>();
public final boolean committer;
public final Set<SharedMemory> sharedMemory = new HashSet<>();
@@ -224,26 +224,21 @@ public class TransactionalTopologyBuilder {
private class SpoutDeclarerImpl extends BaseConfigurationDeclarer<SpoutDeclarer> implements SpoutDeclarer {
@Override
public SpoutDeclarer addConfigurations(Map<String, Object> conf) {
- _spoutConfs.add(conf);
+ if (conf != null && !conf.isEmpty()) {
+ spoutConf.putAll(conf);
+ }
return this;
}
@Override
public Map getRASConfiguration() {
- for(Map<String, Object> conf : _spoutConfs) {
- if (conf.containsKey(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP)) {
- return conf;
- }
- }
- Map<String, Object> newConf = new HashMap<>();
- newConf.put(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, new HashMap());
- _spoutConfs.add(newConf);
- return newConf;
+ //TODO this should be imutable
+ return spoutConf;
}
@Override
public SpoutDeclarer addSharedMemory(SharedMemory request) {
- _spoutSharedMemory.add(request);
+ spoutSharedMemory.add(request);
return this;
}
@@ -263,10 +258,10 @@ public class TransactionalTopologyBuilder {
}
private static class BoltDeclarerImpl extends BaseConfigurationDeclarer<BoltDeclarer> implements BoltDeclarer {
- Component _component;
+ Component component;
public BoltDeclarerImpl(Component component) {
- _component = component;
+ this.component = component;
}
@Override
@@ -552,31 +547,26 @@ public class TransactionalTopologyBuilder {
}
private void addDeclaration(InputDeclaration declaration) {
- _component.declarations.add(declaration);
+ component.declarations.add(declaration);
}
@Override
public BoltDeclarer addConfigurations(Map<String, Object> conf) {
- _component.componentConfs.add(conf);
+ if (conf != null && !conf.isEmpty()) {
+ component.componentConf.putAll(conf);
+ }
return this;
}
@Override
public Map getRASConfiguration() {
- for(Map<String, Object> conf : _component.componentConfs) {
- if (conf.containsKey(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP)) {
- return conf;
- }
- }
- Map<String, Object> newConf = new HashMap<>();
- newConf.put(Config.TOPOLOGY_COMPONENT_RESOURCES_MAP, new HashMap());
- _component.componentConfs.add(newConf);
- return newConf;
+ //TODO this should be read only
+ return component.componentConf;
}
@Override
public BoltDeclarer addSharedMemory(SharedMemory request) {
- _component.sharedMemory.add(request);
+ component.sharedMemory.add(request);
return this;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/96c461d0/storm-client/src/jvm/org/apache/storm/trident/topology/MasterBatchCoordinator.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/trident/topology/MasterBatchCoordinator.java b/storm-client/src/jvm/org/apache/storm/trident/topology/MasterBatchCoordinator.java
index e54d0e2..3417112 100644
--- a/storm-client/src/jvm/org/apache/storm/trident/topology/MasterBatchCoordinator.java
+++ b/storm-client/src/jvm/org/apache/storm/trident/topology/MasterBatchCoordinator.java
@@ -308,7 +308,7 @@ public class MasterBatchCoordinator extends BaseRichSpout {
", _maxTransactionActive=" + _maxTransactionActive +
", _coordinators=" + _coordinators +
", _managedSpoutIds=" + _managedSpoutIds +
- ", _spouts=" + _spouts +
+ ", spouts=" + _spouts +
", _throttler=" + _throttler +
", _active=" + _active +
"}";
http://git-wip-us.apache.org/repos/asf/storm/blob/96c461d0/storm-client/src/jvm/org/apache/storm/trident/topology/TridentTopologyBuilder.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/trident/topology/TridentTopologyBuilder.java b/storm-client/src/jvm/org/apache/storm/trident/topology/TridentTopologyBuilder.java
index be36b4e..274e913 100644
--- a/storm-client/src/jvm/org/apache/storm/trident/topology/TridentTopologyBuilder.java
+++ b/storm-client/src/jvm/org/apache/storm/trident/topology/TridentTopologyBuilder.java
@@ -51,8 +51,8 @@ import org.apache.storm.trident.topology.TridentBoltExecutor.CoordType;
// based on transactional topologies
public class TridentTopologyBuilder {
- Map<GlobalStreamId, String> _batchIds = new HashMap();
- Map<String, TransactionalSpoutComponent> _spouts = new HashMap();
+ Map<GlobalStreamId, String> batchIds = new HashMap();
+ Map<String, TransactionalSpoutComponent> spouts = new HashMap();
Map<String, SpoutComponent> _batchPerTupleSpouts = new HashMap();
Map<String, Component> _bolts = new HashMap();
@@ -76,7 +76,7 @@ public class TridentTopologyBuilder {
markBatchGroups(id, batchGroups);
TransactionalSpoutComponent c = new TransactionalSpoutComponent(spout, streamName, parallelism, txStateId, batchGroup);
- _spouts.put(id, c);
+ spouts.put(id, c);
return new SpoutDeclarerImpl(c);
}
@@ -104,8 +104,8 @@ public class TridentTopologyBuilder {
}
Map<GlobalStreamId, String> fleshOutStreamBatchIds(boolean includeCommitStream) {
- Map<GlobalStreamId, String> ret = new HashMap<>(_batchIds);
- Set<String> allBatches = new HashSet(_batchIds.values());
+ Map<GlobalStreamId, String> ret = new HashMap<>(batchIds);
+ Set<String> allBatches = new HashSet(batchIds.values());
for(String b: allBatches) {
ret.put(new GlobalStreamId(masterCoordinator(b), MasterBatchCoordinator.BATCH_STREAM_ID), b);
if(includeCommitStream) {
@@ -115,16 +115,16 @@ public class TridentTopologyBuilder {
// and is just a metadata tuple to assist in cleanup, should not trigger batch tracking
}
- for(String id: _spouts.keySet()) {
- TransactionalSpoutComponent c = _spouts.get(id);
+ for(String id: spouts.keySet()) {
+ TransactionalSpoutComponent c = spouts.get(id);
if(c.batchGroupId!=null) {
ret.put(new GlobalStreamId(spoutCoordinator(id), MasterBatchCoordinator.BATCH_STREAM_ID), c.batchGroupId);
}
}
//this takes care of setting up coord streams for spouts and bolts
- for(GlobalStreamId s: _batchIds.keySet()) {
- String b = _batchIds.get(s);
+ for(GlobalStreamId s: batchIds.keySet()) {
+ String b = batchIds.get(s);
ret.put(new GlobalStreamId(s.get_componentId(), TridentBoltExecutor.COORD_STREAM(b)), b);
}
@@ -139,8 +139,8 @@ public class TridentTopologyBuilder {
Map<String, List<String>> batchesToCommitIds = new HashMap<>();
Map<String, List<ITridentSpout>> batchesToSpouts = new HashMap<>();
- for(String id: _spouts.keySet()) {
- TransactionalSpoutComponent c = _spouts.get(id);
+ for(String id: spouts.keySet()) {
+ TransactionalSpoutComponent c = spouts.get(id);
if(c.spout instanceof IRichSpout) {
//TODO: wrap this to set the stream name
@@ -273,7 +273,7 @@ public class TridentTopologyBuilder {
private void markBatchGroups(String component, Map<String, String> batchGroups) {
for(Map.Entry<String, String> entry: batchGroups.entrySet()) {
- _batchIds.put(new GlobalStreamId(component, entry.getKey()), entry.getValue());
+ batchIds.put(new GlobalStreamId(component, entry.getKey()), entry.getValue());
}
}
@@ -336,7 +336,7 @@ public class TridentTopologyBuilder {
Map<String, Set<String>> getBoltBatchToComponentSubscriptions(String id) {
Map<String, Set<String>> ret = new HashMap();
for(GlobalStreamId s: getBoltSubscriptionStreams(id)) {
- String b = _batchIds.get(s);
+ String b = batchIds.get(s);
if(!ret.containsKey(b)) ret.put(b, new HashSet());
ret.get(b).add(s.get_componentId());
}
[4/4] storm git commit: Merge branch 'STORM-2805' of
https://github.com/revans2/incubator-storm into STORM-2805
Posted by bo...@apache.org.
Merge branch 'STORM-2805' of https://github.com/revans2/incubator-storm into STORM-2805
STORM-2805: Clean up confs in TopologyBuilders
This closes #2419
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/9d52435c
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/9d52435c
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/9d52435c
Branch: refs/heads/master
Commit: 9d52435c5b5378bc4a618731ef5ecf573f07294a
Parents: 8e4a146 14cf5e6
Author: Robert Evans <ev...@yahoo-inc.com>
Authored: Mon Nov 20 08:12:35 2017 -0600
Committer: Robert Evans <ev...@yahoo-inc.com>
Committed: Mon Nov 20 08:12:35 2017 -0600
----------------------------------------------------------------------
storm-client/pom.xml | 2 +-
.../coordination/BatchSubtopologyBuilder.java | 70 +++++----
.../storm/drpc/LinearDRPCTopologyBuilder.java | 66 ++++-----
.../topology/BaseConfigurationDeclarer.java | 9 --
.../ComponentConfigurationDeclarer.java | 7 -
.../apache/storm/topology/TopologyBuilder.java | 59 ++++----
.../TransactionalTopologyBuilder.java | 118 +++++++--------
.../topology/TridentTopologyBuilder.java | 145 +++++++++----------
storm-server/pom.xml | 2 +-
.../apache/storm/scheduler/TopologyDetails.java | 7 +-
.../storm/scheduler/resource/ResourceUtils.java | 8 +-
.../TestGenericResourceAwareStrategy.java | 112 ++++++++------
12 files changed, 292 insertions(+), 313 deletions(-)
----------------------------------------------------------------------