@Test
public void componentParallelismCouldBeSetThroughConfig() {

  StormBenchmark benchmark = new SOL();
  Config config = new Config();
  config.put(SOL.SPOUT_NUM, 4);
  config.put(SOL.BOLT_NUM, 3);
  config.put(SOL.TOPOLOGY_LEVEL, 3);
  StormTopology topology = benchmark.getTopology(config);
  assertThat(topology).isNotNull();
  TestUtils.verifyParallelism(Utils.getComponentCommon(topology, SOL.SPOUT_ID), 4);
  TestUtils.verifyParallelism(Utils.getComponentCommon(topology, SOL.BOLT_ID + "1"), 3);
  TestUtils.verifyParallelism(Utils.getComponentCommon(topology, SOL.BOLT_ID + "2"), 3);
}