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:26 UTC
[2/4] storm git commit: STORM-2805: Clean up confs in TopologyBuilders
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());
}