Grep extends StormBenchmark { @Override public StormTopology getTopology(Config config) { final int spoutNum = BenchmarkUtils.getInt(config, SPOUT_NUM, DEFAULT_SPOUT_NUM); final int matBoltNum = BenchmarkUtils.getInt(config, FM_NUM, DEFAULT_MAT_BOLT_NUM); final int cntBoltNum = BenchmarkUtils.getInt(config, CM_NUM, DEFAULT_CNT_BOLT_NUM); final String ptnString = (String) Utils.get(config, PATTERN_STRING, DEFAULT_PATTERN_STR); spout = new KafkaSpout(KafkaUtils.getSpoutConfig(config, new SchemeAsMultiScheme(new StringScheme()))); TopologyBuilder builder = new TopologyBuilder(); builder.setSpout(SPOUT_ID, spout, spoutNum); builder.setBolt(FM_ID, new FindMatchingSentence(ptnString), matBoltNum) .localOrShuffleGrouping(SPOUT_ID); builder.setBolt(CM_ID, new CountMatchingSentence(), cntBoltNum) .fieldsGrouping(FM_ID, new Fields(FindMatchingSentence.FIELDS)); return builder.createTopology(); }  @Override StormTopology getTopology(Config config); static final String SPOUT_ID; static final String SPOUT_NUM; static final String FM_ID; static final String FM_NUM; static final String CM_ID; static final String CM_NUM; static final String PATTERN_STRING; static final String DEFAULT_PATTERN_STR; static final int DEFAULT_SPOUT_NUM; static final int DEFAULT_MAT_BOLT_NUM; static final int DEFAULT_CNT_BOLT_NUM; }
@Test public void componentParallelismCouldBeSetThroughConfig() { StormBenchmark benchmark = new Grep(); Config config = new Config(); config.put(Grep.SPOUT_NUM, 3); config.put(Grep.FM_NUM, 4); config.put(Grep.CM_NUM, 5); StormTopology topology = benchmark.getTopology(config); assertThat(topology).isNotNull(); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, Grep.SPOUT_ID), 3); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, Grep.FM_ID), 4); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, Grep.CM_ID), 5); }