test_class.file: storm/storm-benchmark/src/test/java/storm/benchmark/benchmarks/RollingSortTest.java
test_class.identifier: RollingSortTest
test_case.body: @Test
    public void componentParallelismCouldBeSetThroughConfig() {
      StormBenchmark benchmark = new RollingSort();
      Config config = new Config();
      config.put(RollingSort.SPOUT_NUM, 4);
      config.put(RollingSort.SORT_BOLT_NUM, 5);
  
      StormTopology topology = benchmark.getTopology(config);
      assertThat(topology).isNotNull();
      TestUtils.verifyParallelism(Utils.getComponentCommon(topology, RollingSort.SPOUT_ID), 4);
      TestUtils.verifyParallelism(Utils.getComponentCommon(topology, RollingSort.SORT_BOLT_ID), 5);
    }
test_case.class_method_signature: RollingSortTest.componentParallelismCouldBeSetThroughConfig()
test_case.full_signature: @Test public void componentParallelismCouldBeSetThroughConfig()
test_case.identifier: componentParallelismCouldBeSetThroughConfig
test_case.invocations: put put getTopology isNotNull assertThat verifyParallelism getComponentCommon verifyParallelism getComponentCommon
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void componentParallelismCouldBeSetThroughConfig()
focal_class.declarator: LOG = Logger.getLogger(RollingSort.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger LOG = Logger.getLogger(RollingSort.class);
focal_class.type: Logger
focal_class.var_name: LOG
focal_class.declarator: SPOUT_ID = "spout"
focal_class.modifier: public static final
focal_class.original_string: public static final String SPOUT_ID = "spout";
focal_class.type: String
focal_class.var_name: SPOUT_ID
focal_class.declarator: SPOUT_NUM = "component.spout_num"
focal_class.modifier: public static final
focal_class.original_string: public static final String SPOUT_NUM = "component.spout_num";
focal_class.type: String
focal_class.var_name: SPOUT_NUM
focal_class.declarator: SORT_BOLT_ID ="sort"
focal_class.modifier: public static final
focal_class.original_string: public static final String SORT_BOLT_ID ="sort";
focal_class.type: String
focal_class.var_name: SORT_BOLT_ID
focal_class.declarator: SORT_BOLT_NUM = "component.sort_bolt_num"
focal_class.modifier: public static final
focal_class.original_string: public static final String SORT_BOLT_NUM = "component.sort_bolt_num";
focal_class.type: String
focal_class.var_name: SORT_BOLT_NUM
focal_class.declarator: DEFAULT_SPOUT_NUM = 4
focal_class.modifier: public static final
focal_class.original_string: public static final int DEFAULT_SPOUT_NUM = 4;
focal_class.type: int
focal_class.var_name: DEFAULT_SPOUT_NUM
focal_class.declarator: DEFAULT_SORT_BOLT_NUM = 8
focal_class.modifier: public static final
focal_class.original_string: public static final int DEFAULT_SORT_BOLT_NUM = 8;
focal_class.type: int
focal_class.var_name: DEFAULT_SORT_BOLT_NUM
focal_class.declarator: spout
focal_class.modifier: private
focal_class.original_string: private IRichSpout spout;
focal_class.type: IRichSpout
focal_class.var_name: spout
focal_class.file: storm/storm-benchmark/src/main/java/storm/benchmark/benchmarks/RollingSort.java
focal_class.identifier: RollingSort
focal_class.class_method_signature: RollingSort.getTopology(Config config)
focal_class.full_signature: @Override public StormTopology getTopology(Config config)
focal_class.identifier: getTopology
focal_class.modifiers: @Override public
focal_class.parameters: (Config config)
focal_class.return: StormTopology
focal_class.signature: StormTopology getTopology(Config config)
focal_class.superclass: extends StormBenchmark
focal_method.body: @Override
    public StormTopology getTopology(Config config) {
      final int spoutNum = BenchmarkUtils.getInt(config, SPOUT_NUM, DEFAULT_SPOUT_NUM);
      final int boltNum =  BenchmarkUtils.getInt(config, SORT_BOLT_NUM, DEFAULT_SORT_BOLT_NUM);
      final int msgSize = BenchmarkUtils.getInt(config, RandomMessageSpout.MESSAGE_SIZE,
              RandomMessageSpout.DEFAULT_MESSAGE_SIZE);
      final int chunkSize = BenchmarkUtils.getInt(config, SortBolt.CHUNK_SIZE,
              SortBolt.DEFAULT_CHUNK_SIZE);
      final int emitFreq = BenchmarkUtils.getInt(config, SortBolt.EMIT_FREQ,
              SortBolt.DEFAULT_EMIT_FREQ);
      spout = new RandomMessageSpout(msgSize, BenchmarkUtils.ifAckEnabled(config));
      TopologyBuilder builder = new TopologyBuilder();
      builder.setSpout(SPOUT_ID, spout, spoutNum);
      builder.setBolt(SORT_BOLT_ID, new SortBolt(emitFreq, chunkSize), boltNum)
              .localOrShuffleGrouping(SPOUT_ID);
      return builder.createTopology();
    }
focal_method.class_method_signature: RollingSort.getTopology(Config config)
focal_method.full_signature: @Override public StormTopology getTopology(Config config)
focal_method.identifier: getTopology
focal_method.invocations: getInt getInt getInt getInt getInt ifAckEnabled setSpout localOrShuffleGrouping setBolt createTopology
focal_method.modifiers: @Override public
focal_method.parameters: (Config config)
focal_method.return: StormTopology
focal_method.signature: StormTopology getTopology(Config config)