test_class.file: storm/storm-benchmark/src/test/java/storm/benchmark/benchmarks/GrepTest.java
test_class.identifier: GrepTest
test_case.body: @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);
    }
test_case.class_method_signature: GrepTest.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: LOG = Logger.getLogger(Grep.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger LOG = Logger.getLogger(Grep.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: FM_ID = "find"
focal_class.modifier: public static final
focal_class.original_string: public static final String FM_ID = "find";
focal_class.type: String
focal_class.var_name: FM_ID
focal_class.declarator: FM_NUM = "component.find_bolt_num"
focal_class.modifier: public static final
focal_class.original_string: public static final String FM_NUM = "component.find_bolt_num";
focal_class.type: String
focal_class.var_name: FM_NUM
focal_class.declarator: CM_ID = "count"
focal_class.modifier: public static final
focal_class.original_string: public static final String CM_ID = "count";
focal_class.type: String
focal_class.var_name: CM_ID
focal_class.declarator: CM_NUM = "component.count_bolt_num"
focal_class.modifier: public static final
focal_class.original_string: public static final String CM_NUM = "component.count_bolt_num";
focal_class.type: String
focal_class.var_name: CM_NUM
focal_class.declarator: PATTERN_STRING = "pattern_string"
focal_class.modifier: public static final
focal_class.original_string: public static final String PATTERN_STRING = "pattern_string";
focal_class.type: String
focal_class.var_name: PATTERN_STRING
focal_class.declarator: DEFAULT_PATTERN_STR = "string"
focal_class.modifier: public static final
focal_class.original_string: public static final String DEFAULT_PATTERN_STR = "string";
focal_class.type: String
focal_class.var_name: DEFAULT_PATTERN_STR
focal_class.declarator: DEFAULT_SPOUT_NUM = 5
focal_class.modifier: public static final
focal_class.original_string: public static final int DEFAULT_SPOUT_NUM = 5;
focal_class.type: int
focal_class.var_name: DEFAULT_SPOUT_NUM
focal_class.declarator: DEFAULT_MAT_BOLT_NUM = 8
focal_class.modifier: public static final
focal_class.original_string: public static final int DEFAULT_MAT_BOLT_NUM = 8;
focal_class.type: int
focal_class.var_name: DEFAULT_MAT_BOLT_NUM
focal_class.declarator: DEFAULT_CNT_BOLT_NUM = 4
focal_class.modifier: public static final
focal_class.original_string: public static final int DEFAULT_CNT_BOLT_NUM = 4;
focal_class.type: int
focal_class.var_name: DEFAULT_CNT_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/Grep.java
focal_class.identifier: Grep
focal_class.class_method_signature: Grep.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 matBoltNum = BenchmarkUtils.getInt(config, FM_NUM, DEFAULT_MAT_BOLT_NUM);
      final int cntBoltNum = BenchmarkUtils.getInt(config, CM_NUM, DEFAULT_CNT_BOLT_NUM);
      final String ptnString = (String) Utils.get(config, PATTERN_STRING, DEFAULT_PATTERN_STR);
  
      spout = new KafkaSpout(KafkaUtils.getSpoutConfig(config, new SchemeAsMultiScheme(new StringScheme())));
  
      TopologyBuilder builder = new TopologyBuilder();
      builder.setSpout(SPOUT_ID, spout, spoutNum);
      builder.setBolt(FM_ID, new FindMatchingSentence(ptnString), matBoltNum)
              .localOrShuffleGrouping(SPOUT_ID);
      builder.setBolt(CM_ID, new CountMatchingSentence(), cntBoltNum)
              .fieldsGrouping(FM_ID, new Fields(FindMatchingSentence.FIELDS));
  
      return builder.createTopology();
    }
focal_method.class_method_signature: Grep.getTopology(Config config)
focal_method.full_signature: @Override public StormTopology getTopology(Config config)
focal_method.identifier: getTopology
focal_method.invocations: getInt getInt getInt get getSpoutConfig 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)