FileReadWordCount extends WordCount { @Override public StormTopology getTopology(Config config) { spout = new FileReadSpout(BenchmarkUtils.ifAckEnabled(config)); return super.getTopology(config); }  @Override StormTopology getTopology(Config config);  }
@Test public void componentParallelismCouldBeSetThroughConfig() { StormBenchmark benchmark = new FileReadWordCount(); Config config = new Config(); config.put(WordCount.SPOUT_NUM, 3); config.put(WordCount.SPLIT_NUM, 4); config.put(WordCount.COUNT_NUM, 5); StormTopology topology = benchmark.getTopology(config); assertThat(topology).isNotNull(); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, WordCount.SPOUT_ID), 3); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, WordCount.SPLIT_ID), 4); TestUtils.verifyParallelism(Utils.getComponentCommon(topology, WordCount.COUNT_ID), 5); }