KafkaProducer implements IProducer { @Override public StormTopology getTopology(Config config) { config.putAll(getKafkaConfig(config)); final int spoutNum = BenchmarkUtils.getInt(config , SPOUT_NUM, DEFAULT_SPOUT_NUM); final int boltNum = BenchmarkUtils.getInt(config, BOLT_NUM, DEFAULT_BOLT_NUM); TopologyBuilder builder = new TopologyBuilder(); builder.setSpout(SPOUT_ID, spout, spoutNum); builder.setBolt(BOLT_ID, bolt, boltNum).localOrShuffleGrouping(SPOUT_ID); return builder.createTopology(); }  @Override StormTopology getTopology(Config config); IRichSpout getSpout(); static final String SPOUT_ID; static final String SPOUT_NUM; static final String BOLT_ID; static final String BOLT_NUM; static final String BROKER_LIST; static final String TOPIC; static final int DEFAULT_SPOUT_NUM; static final int DEFAULT_BOLT_NUM; }
@Test(dataProvider = "getKafkaProducer") public void componentParallelismCouldBeSetThroughConfig(KafkaProducer producer) { Config config = new Config(); config.put(KafkaProducer.SPOUT_NUM, 4); config.put(KafkaProducer.BOLT_NUM, 3); StormTopology topology = producer.getTopology(config); assertThat(topology).isNotNull(); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, KafkaProducer.SPOUT_ID), 4); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, KafkaProducer.BOLT_ID), 3); }