test_class.file: storm/storm-benchmark/src/test/java/storm/benchmark/benchmarks/RollingCountTest.java
test_class.identifier: RollingCountTest
test_case.body: @Test
    public void componentParallelismCouldBeSetThroughConfig() {
      StormBenchmark benchmark = new RollingCount();
      Config config = new Config();
      config.put(RollingCount.SPOUT_NUM, 4);
      config.put(RollingCount.SPLIT_NUM, 5);
      config.put(RollingCount.COUNTER_NUM, 3);
  
      StormTopology topology = benchmark.getTopology(config);
      assertThat(topology).isNotNull();
      TestUtils.verifyParallelism(Utils.getComponentCommon(topology, RollingCount.SPOUT_ID), 4);
      TestUtils.verifyParallelism(Utils.getComponentCommon(topology, RollingCount.SPLIT_ID), 5);
      TestUtils.verifyParallelism(Utils.getComponentCommon(topology, RollingCount.COUNTER_ID), 3);
    }
test_case.class_method_signature: RollingCountTest.componentParallelismCouldBeSetThroughConfig()
test_case.full_signature: @Test public void componentParallelismCouldBeSetThroughConfig()
test_case.identifier: componentParallelismCouldBeSetThroughConfig
test_case.invocations: put put put getTopology isNotNull assertThat verifyParallelism getComponentCommon verifyParallelism getComponentCommon verifyParallelism getComponentCommon
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void componentParallelismCouldBeSetThroughConfig()
focal_class.declarator: WINDOW_LENGTH = "window.length"
focal_class.modifier: private static final
focal_class.original_string: private static final String WINDOW_LENGTH = "window.length";
focal_class.type: String
focal_class.var_name: WINDOW_LENGTH
focal_class.declarator: EMIT_FREQ = "emit.frequency"
focal_class.modifier: private static final
focal_class.original_string: private static final String EMIT_FREQ = "emit.frequency";
focal_class.type: String
focal_class.var_name: EMIT_FREQ
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: SPLIT_ID = "split"
focal_class.modifier: public static final
focal_class.original_string: public static final String SPLIT_ID = "split";
focal_class.type: String
focal_class.var_name: SPLIT_ID
focal_class.declarator: SPLIT_NUM = "component.split_bolt_num"
focal_class.modifier: public static final
focal_class.original_string: public static final String SPLIT_NUM = "component.split_bolt_num";
focal_class.type: String
focal_class.var_name: SPLIT_NUM
focal_class.declarator: COUNTER_ID = "rolling_count"
focal_class.modifier: public static final
focal_class.original_string: public static final String COUNTER_ID = "rolling_count";
focal_class.type: String
focal_class.var_name: COUNTER_ID
focal_class.declarator: COUNTER_NUM = "component.rolling_count_bolt_num"
focal_class.modifier: public static final
focal_class.original_string: public static final String COUNTER_NUM = "component.rolling_count_bolt_num";
focal_class.type: String
focal_class.var_name: COUNTER_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_SP_BOLT_NUM = 8
focal_class.modifier: public static final
focal_class.original_string: public static final int DEFAULT_SP_BOLT_NUM = 8;
focal_class.type: int
focal_class.var_name: DEFAULT_SP_BOLT_NUM
focal_class.declarator: DEFAULT_RC_BOLT_NUM = 8
focal_class.modifier: public static final
focal_class.original_string: public static final int DEFAULT_RC_BOLT_NUM = 8;
focal_class.type: int
focal_class.var_name: DEFAULT_RC_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/RollingCount.java
focal_class.identifier: RollingCount
focal_class.class_method_signature: RollingCount.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 spBoltNum = BenchmarkUtils.getInt(config, SPLIT_NUM, DEFAULT_SP_BOLT_NUM);
      final int rcBoltNum = BenchmarkUtils.getInt(config, COUNTER_NUM, DEFAULT_RC_BOLT_NUM);
      final int windowLength = BenchmarkUtils.getInt(config, WINDOW_LENGTH,
              RollingBolt.DEFAULT_SLIDING_WINDOW_IN_SECONDS);
      final int emitFreq = BenchmarkUtils.getInt(config, EMIT_FREQ,
              RollingBolt.DEFAULT_EMIT_FREQUENCY_IN_SECONDS);
  
      spout = new FileReadSpout(BenchmarkUtils.ifAckEnabled(config));
  
      TopologyBuilder builder = new TopologyBuilder();
  
      builder.setSpout(SPOUT_ID, spout, spoutNum);
      builder.setBolt(SPLIT_ID, new WordCount.SplitSentence(), spBoltNum)
              .localOrShuffleGrouping(SPOUT_ID);
      builder.setBolt(COUNTER_ID, new RollingCountBolt(windowLength, emitFreq), rcBoltNum)
              .fieldsGrouping(SPLIT_ID, new Fields(WordCount.SplitSentence.FIELDS));
      return builder.createTopology();
    }
focal_method.class_method_signature: RollingCount.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 fieldsGrouping setBolt createTopology
focal_method.modifiers: @Override public
focal_method.parameters: (Config config)
focal_method.return: StormTopology
focal_method.signature: StormTopology getTopology(Config config)