SOL extends StormBenchmark { @Override public StormTopology getTopology(Config config) { final int numLevels = BenchmarkUtils.getInt(config, TOPOLOGY_LEVEL, DEFAULT_NUM_LEVELS); final int msgSize = BenchmarkUtils.getInt(config, RandomMessageSpout.MESSAGE_SIZE, RandomMessageSpout.DEFAULT_MESSAGE_SIZE); final int spoutNum = BenchmarkUtils.getInt(config, SPOUT_NUM, DEFAULT_SPOUT_NUM); final int boltNum = BenchmarkUtils.getInt(config, BOLT_NUM, DEFAULT_BOLT_NUM); spout = new RandomMessageSpout(msgSize, BenchmarkUtils.ifAckEnabled(config)); TopologyBuilder builder = new TopologyBuilder(); builder.setSpout(SPOUT_ID, spout, spoutNum); builder.setBolt(BOLT_ID + 1, new ConstBolt(), boltNum) .shuffleGrouping(SPOUT_ID); for (int levelNum = 2; levelNum <= numLevels - 1; levelNum++) { builder.setBolt(BOLT_ID + levelNum, new ConstBolt(), boltNum) .shuffleGrouping(BOLT_ID + (levelNum - 1)); } return builder.createTopology(); }  @Override StormTopology getTopology(Config config); @Override IMetricsCollector getMetricsCollector(Config config, StormTopology topology); static final String TOPOLOGY_LEVEL; static final String SPOUT_ID; static final String SPOUT_NUM; static final String BOLT_ID; static final String BOLT_NUM; static final int DEFAULT_NUM_LEVELS; static final int DEFAULT_SPOUT_NUM; static final int DEFAULT_BOLT_NUM; }
@Test public void componentParallelismCouldBeSetThroughConfig() { StormBenchmark benchmark = new SOL(); Config config = new Config(); config.put(SOL.SPOUT_NUM, 4); config.put(SOL.BOLT_NUM, 3); config.put(SOL.TOPOLOGY_LEVEL, 3); StormTopology topology = benchmark.getTopology(config); assertThat(topology).isNotNull(); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, SOL.SPOUT_ID), 4); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, SOL.BOLT_ID + "1"), 3); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, SOL.BOLT_ID + "2"), 3); }