RollingSort extends StormBenchmark { @Override public StormTopology getTopology(Config config) { final int spoutNum = BenchmarkUtils.getInt(config, SPOUT_NUM, DEFAULT_SPOUT_NUM); final int boltNum = BenchmarkUtils.getInt(config, SORT_BOLT_NUM, DEFAULT_SORT_BOLT_NUM); final int msgSize = BenchmarkUtils.getInt(config, RandomMessageSpout.MESSAGE_SIZE, RandomMessageSpout.DEFAULT_MESSAGE_SIZE); final int chunkSize = BenchmarkUtils.getInt(config, SortBolt.CHUNK_SIZE, SortBolt.DEFAULT_CHUNK_SIZE); final int emitFreq = BenchmarkUtils.getInt(config, SortBolt.EMIT_FREQ, SortBolt.DEFAULT_EMIT_FREQ); spout = new RandomMessageSpout(msgSize, BenchmarkUtils.ifAckEnabled(config)); TopologyBuilder builder = new TopologyBuilder(); builder.setSpout(SPOUT_ID, spout, spoutNum); builder.setBolt(SORT_BOLT_ID, new SortBolt(emitFreq, chunkSize), boltNum) .localOrShuffleGrouping(SPOUT_ID); return builder.createTopology(); }  @Override StormTopology getTopology(Config config); static final String SPOUT_ID; static final String SPOUT_NUM; static final String SORT_BOLT_ID; static final String SORT_BOLT_NUM; static final int DEFAULT_SPOUT_NUM; static final int DEFAULT_SORT_BOLT_NUM; }
@Test public void componentParallelismCouldBeSetThroughConfig() { StormBenchmark benchmark = new RollingSort(); Config config = new Config(); config.put(RollingSort.SPOUT_NUM, 4); config.put(RollingSort.SORT_BOLT_NUM, 5); StormTopology topology = benchmark.getTopology(config); assertThat(topology).isNotNull(); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, RollingSort.SPOUT_ID), 4); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, RollingSort.SORT_BOLT_ID), 5); }