@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);
}