RollingCount extends StormBenchmark { @Override public StormTopology getTopology(Config config) { final int spoutNum = BenchmarkUtils.getInt(config, SPOUT_NUM, DEFAULT_SPOUT_NUM); final int spBoltNum = BenchmarkUtils.getInt(config, SPLIT_NUM, DEFAULT_SP_BOLT_NUM); final int rcBoltNum = BenchmarkUtils.getInt(config, COUNTER_NUM, DEFAULT_RC_BOLT_NUM); final int windowLength = BenchmarkUtils.getInt(config, WINDOW_LENGTH, RollingBolt.DEFAULT_SLIDING_WINDOW_IN_SECONDS); final int emitFreq = BenchmarkUtils.getInt(config, EMIT_FREQ, RollingBolt.DEFAULT_EMIT_FREQUENCY_IN_SECONDS); spout = new FileReadSpout(BenchmarkUtils.ifAckEnabled(config)); TopologyBuilder builder = new TopologyBuilder(); builder.setSpout(SPOUT_ID, spout, spoutNum); builder.setBolt(SPLIT_ID, new WordCount.SplitSentence(), spBoltNum) .localOrShuffleGrouping(SPOUT_ID); builder.setBolt(COUNTER_ID, new RollingCountBolt(windowLength, emitFreq), rcBoltNum) .fieldsGrouping(SPLIT_ID, new Fields(WordCount.SplitSentence.FIELDS)); return builder.createTopology(); }  @Override StormTopology getTopology(Config config); static final String SPOUT_ID; static final String SPOUT_NUM; static final String SPLIT_ID; static final String SPLIT_NUM; static final String COUNTER_ID; static final String COUNTER_NUM; static final int DEFAULT_SPOUT_NUM; static final int DEFAULT_SP_BOLT_NUM; static final int DEFAULT_RC_BOLT_NUM; }
@Test public void componentParallelismCouldBeSetThroughConfig() { StormBenchmark benchmark = new RollingCount(); Config config = new Config(); config.put(RollingCount.SPOUT_NUM, 4); config.put(RollingCount.SPLIT_NUM, 5); config.put(RollingCount.COUNTER_NUM, 3); StormTopology topology = benchmark.getTopology(config); assertThat(topology).isNotNull(); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, RollingCount.SPOUT_ID), 4); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, RollingCount.SPLIT_ID), 5); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, RollingCount.COUNTER_ID), 3); }