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