UniqueVisitor extends StormBenchmark { @Override public StormTopology getTopology(Config config) { final int spoutNum = BenchmarkUtils.getInt(config, SPOUT_NUM, DEFAULT_SPOUT_NUM); final int pvBoltNum = BenchmarkUtils.getInt(config, VIEW_NUM, DEFAULT_PV_BOLT_NUM); final int uvBoltNum = BenchmarkUtils.getInt(config, UNIQUER_NUM, DEFAULT_UV_BOLT_NUM); final int winLen = BenchmarkUtils.getInt(config, WINDOW_LENGTH, DEFAULT_WINDOW_LENGTH_IN_SEC); final int emitFreq = BenchmarkUtils.getInt(config, EMIT_FREQ, DEFAULT_EMIT_FREQ_IN_SEC); spout = new KafkaSpout(KafkaUtils.getSpoutConfig( config, new SchemeAsMultiScheme(new StringScheme()))); TopologyBuilder builder = new TopologyBuilder(); builder.setSpout(SPOUT_ID, spout, spoutNum); builder.setBolt(VIEW_ID, new PageViewBolt(Item.URL, Item.USER), pvBoltNum) .localOrShuffleGrouping(SPOUT_ID); builder.setBolt(UNIQUER_ID, new UniqueVisitorBolt(winLen, emitFreq), uvBoltNum) .fieldsGrouping(VIEW_ID, new Fields(Item.URL.toString())); return builder.createTopology(); }  @Override StormTopology getTopology(Config config); final static String SPOUT_ID; final static String SPOUT_NUM; final static String VIEW_ID; final static String VIEW_NUM; final static String UNIQUER_ID; final static String UNIQUER_NUM; final static String WINDOW_LENGTH; final static String EMIT_FREQ; static final int DEFAULT_SPOUT_NUM; static final int DEFAULT_PV_BOLT_NUM; static final int DEFAULT_UV_BOLT_NUM; static final int DEFAULT_WINDOW_LENGTH_IN_SEC; static final int DEFAULT_EMIT_FREQ_IN_SEC; }
@Test public void componentParallelismCouldBeSetThroughConfig() { StormBenchmark benchmark = new UniqueVisitor(); Config config = new Config(); config.put(UniqueVisitor.SPOUT_NUM, 3); config.put(UniqueVisitor.VIEW_NUM, 4); config.put(UniqueVisitor.UNIQUER_NUM, 5); StormTopology topology = benchmark.getTopology(config); assertThat(topology).isNotNull(); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, UniqueVisitor.SPOUT_ID), 3); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, UniqueVisitor.VIEW_ID), 4); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, UniqueVisitor.UNIQUER_ID), 5); }