test_class.file: storm/storm-benchmark/src/test/java/storm/benchmark/tools/producer/kafka/KafkaProducerTest.java
test_class.identifier: KafkaProducerTest
test_case.body: @Test(dataProvider = "getKafkaProducer")
    public void componentParallelismCouldBeSetThroughConfig(KafkaProducer producer) {
      Config config = new Config();
      config.put(KafkaProducer.SPOUT_NUM, 4);
      config.put(KafkaProducer.BOLT_NUM, 3);
      StormTopology topology = producer.getTopology(config);
      assertThat(topology).isNotNull();
      TestUtils.verifyParallelism(Utils.getComponentCommon(topology, KafkaProducer.SPOUT_ID), 4);
      TestUtils.verifyParallelism(Utils.getComponentCommon(topology, KafkaProducer.BOLT_ID), 3);
    }
test_case.class_method_signature: KafkaProducerTest.componentParallelismCouldBeSetThroughConfig(KafkaProducer producer)
test_case.full_signature: @Test(dataProvider = "getKafkaProducer") public void componentParallelismCouldBeSetThroughConfig(KafkaProducer producer)
test_case.identifier: componentParallelismCouldBeSetThroughConfig
test_case.invocations: put put getTopology isNotNull assertThat verifyParallelism getComponentCommon verifyParallelism getComponentCommon
test_case.modifiers: @Test(dataProvider = "getKafkaProducer") public
test_case.parameters: (KafkaProducer producer)
test_case.return: void
test_case.signature: void componentParallelismCouldBeSetThroughConfig(KafkaProducer producer)
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: BOLT_ID = "bolt"
focal_class.modifier: public static final
focal_class.original_string: public static final String BOLT_ID = "bolt";
focal_class.type: String
focal_class.var_name: BOLT_ID
focal_class.declarator: BOLT_NUM = "component.bolt_num"
focal_class.modifier: public static final
focal_class.original_string: public static final String BOLT_NUM = "component.bolt_num";
focal_class.type: String
focal_class.var_name: BOLT_NUM
focal_class.declarator: BROKER_LIST = "broker.list"
focal_class.modifier: public static final
focal_class.original_string: public static final String BROKER_LIST = "broker.list";
focal_class.type: String
focal_class.var_name: BROKER_LIST
focal_class.declarator: TOPIC = "topic"
focal_class.modifier: public static final
focal_class.original_string: public static final String TOPIC = "topic";
focal_class.type: String
focal_class.var_name: TOPIC
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_BOLT_NUM = 4
focal_class.modifier: public static final
focal_class.original_string: public static final int DEFAULT_BOLT_NUM = 4;
focal_class.type: int
focal_class.var_name: DEFAULT_BOLT_NUM
focal_class.declarator: spout
focal_class.modifier: protected
focal_class.original_string: protected IRichSpout spout;
focal_class.type: IRichSpout
focal_class.var_name: spout
focal_class.declarator: bolt = new KafkaBolt<String, String>()
focal_class.modifier: protected final
focal_class.original_string: protected final IRichBolt bolt = new KafkaBolt<String, String>();
focal_class.type: IRichBolt
focal_class.var_name: bolt
focal_class.file: storm/storm-benchmark/src/main/java/storm/benchmark/tools/producer/kafka/KafkaProducer.java
focal_class.identifier: KafkaProducer
focal_class.interfaces: implements IProducer
focal_class.class_method_signature: KafkaProducer.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.class_method_signature: KafkaProducer.getSpout()
focal_class.full_signature: public IRichSpout getSpout()
focal_class.identifier: getSpout
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: IRichSpout
focal_class.signature: IRichSpout getSpout()
focal_class.class_method_signature: KafkaProducer.getKafkaConfig(Map options)
focal_class.full_signature: private Map getKafkaConfig(Map options)
focal_class.identifier: getKafkaConfig
focal_class.modifiers: private
focal_class.parameters: (Map options)
focal_class.return: Map
focal_class.signature: Map getKafkaConfig(Map options)
focal_method.body: @Override
    public StormTopology getTopology(Config config) {
      config.putAll(getKafkaConfig(config));
  
      final int spoutNum = BenchmarkUtils.getInt(config , SPOUT_NUM, DEFAULT_SPOUT_NUM);
      final int boltNum = BenchmarkUtils.getInt(config, BOLT_NUM, DEFAULT_BOLT_NUM);
  
      TopologyBuilder builder = new TopologyBuilder();
      builder.setSpout(SPOUT_ID, spout, spoutNum);
      builder.setBolt(BOLT_ID, bolt, boltNum).localOrShuffleGrouping(SPOUT_ID);
      return builder.createTopology();
    }
focal_method.class_method_signature: KafkaProducer.getTopology(Config config)
focal_method.full_signature: @Override public StormTopology getTopology(Config config)
focal_method.identifier: getTopology
focal_method.invocations: putAll getKafkaConfig getInt getInt 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)