test_class.declarator: TEST_TOPIC = "test-topic"
test_class.modifier: private static final
test_class.original_string: private static final String TEST_TOPIC = "test-topic";
test_class.type: String
test_class.var_name: TEST_TOPIC
test_class.declarator: broker
test_class.modifier: private
test_class.original_string: private KafkaTestBroker broker;
test_class.type: KafkaTestBroker
test_class.var_name: broker
test_class.declarator: bolt
test_class.modifier: private
test_class.original_string: private KafkaBolt bolt;
test_class.type: KafkaBolt
test_class.var_name: bolt
test_class.declarator: config = new Config()
test_class.modifier: private
test_class.original_string: private Config config = new Config();
test_class.type: Config
test_class.var_name: config
test_class.declarator: kafkaConfig
test_class.modifier: private
test_class.original_string: private KafkaConfig kafkaConfig;
test_class.type: KafkaConfig
test_class.var_name: kafkaConfig
test_class.declarator: simpleConsumer
test_class.modifier: private
test_class.original_string: private SimpleConsumer simpleConsumer;
test_class.type: SimpleConsumer
test_class.var_name: simpleConsumer
test_class.declarator: collector
test_class.modifier: @Mock
      private
test_class.original_string: @Mock
      private IOutputCollector collector;
test_class.type: IOutputCollector
test_class.var_name: collector
test_class.file: storm/storm-app/src/test/java/storm/kafka/bolt/KafkaBoltTest.java
test_class.identifier: KafkaBoltTest
test_case.body: @Test
      public void executeWithByteArrayKeyAndMessage() {
          bolt = generateDefaultSerializerBolt();
          String keyString = "test-key";
          String messageString = "test-message";
          byte[] key = keyString.getBytes();
          byte[] message = messageString.getBytes();
          Tuple tuple = generateTestTuple(key, message);
          bolt.execute(tuple);
          verify(collector).ack(tuple);
          verifyMessage(keyString, messageString);
      }
test_case.class_method_signature: KafkaBoltTest.executeWithByteArrayKeyAndMessage()
test_case.full_signature: @Test public void executeWithByteArrayKeyAndMessage()
test_case.identifier: executeWithByteArrayKeyAndMessage
test_case.invocations: generateDefaultSerializerBolt getBytes getBytes generateTestTuple execute ack verify verifyMessage
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void executeWithByteArrayKeyAndMessage()
focal_class.declarator: LOG = LoggerFactory.getLogger(KafkaBolt.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger LOG = LoggerFactory.getLogger(KafkaBolt.class);
focal_class.type: Logger
focal_class.var_name: LOG
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: KAFKA_BROKER_PROPERTIES = "kafka.broker.properties"
focal_class.modifier: public static final
focal_class.original_string: public static final String KAFKA_BROKER_PROPERTIES = "kafka.broker.properties";
focal_class.type: String
focal_class.var_name: KAFKA_BROKER_PROPERTIES
focal_class.declarator: producer
focal_class.modifier: private
focal_class.original_string: private Producer<K, V> producer;
focal_class.type: Producer<K, V>
focal_class.var_name: producer
focal_class.declarator: collector
focal_class.modifier: private
focal_class.original_string: private OutputCollector collector;
focal_class.type: OutputCollector
focal_class.var_name: collector
focal_class.declarator: mapper
focal_class.modifier: private
focal_class.original_string: private TupleToKafkaMapper<K,V> mapper;
focal_class.type: TupleToKafkaMapper<K,V>
focal_class.var_name: mapper
focal_class.declarator: topicSelector
focal_class.modifier: private
focal_class.original_string: private KafkaTopicSelector topicSelector;
focal_class.type: KafkaTopicSelector
focal_class.var_name: topicSelector
focal_class.file: storm/storm-app/src/main/java/storm/kafka/bolt/KafkaBolt.java
focal_class.identifier: KafkaBolt
focal_class.class_method_signature: KafkaBolt.withTupleToKafkaMapper(TupleToKafkaMapper<K,V> mapper)
focal_class.full_signature: public KafkaBolt<K,V> withTupleToKafkaMapper(TupleToKafkaMapper<K,V> mapper)
focal_class.identifier: withTupleToKafkaMapper
focal_class.modifiers: public
focal_class.parameters: (TupleToKafkaMapper<K,V> mapper)
focal_class.return: KafkaBolt<K,V>
focal_class.signature: KafkaBolt<K,V> withTupleToKafkaMapper(TupleToKafkaMapper<K,V> mapper)
focal_class.class_method_signature: KafkaBolt.withTopicSelector(KafkaTopicSelector selector)
focal_class.full_signature: public KafkaBolt<K,V> withTopicSelector(KafkaTopicSelector selector)
focal_class.identifier: withTopicSelector
focal_class.modifiers: public
focal_class.parameters: (KafkaTopicSelector selector)
focal_class.return: KafkaBolt<K,V>
focal_class.signature: KafkaBolt<K,V> withTopicSelector(KafkaTopicSelector selector)
focal_class.class_method_signature: KafkaBolt.prepare(Map stormConf, TopologyContext context, OutputCollector collector)
focal_class.full_signature: @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector)
focal_class.identifier: prepare
focal_class.modifiers: @Override public
focal_class.parameters: (Map stormConf, TopologyContext context, OutputCollector collector)
focal_class.return: void
focal_class.signature: void prepare(Map stormConf, TopologyContext context, OutputCollector collector)
focal_class.class_method_signature: KafkaBolt.execute(Tuple input)
focal_class.full_signature: @Override public void execute(Tuple input)
focal_class.identifier: execute
focal_class.modifiers: @Override public
focal_class.parameters: (Tuple input)
focal_class.return: void
focal_class.signature: void execute(Tuple input)
focal_class.class_method_signature: KafkaBolt.declareOutputFields(OutputFieldsDeclarer declarer)
focal_class.full_signature: @Override public void declareOutputFields(OutputFieldsDeclarer declarer)
focal_class.identifier: declareOutputFields
focal_class.modifiers: @Override public
focal_class.parameters: (OutputFieldsDeclarer declarer)
focal_class.return: void
focal_class.signature: void declareOutputFields(OutputFieldsDeclarer declarer)
focal_class.superclass: extends BaseRichBolt
focal_method.body: @Override
      public void execute(Tuple input) {
          K key = null;
          V message = null;
          String topic = null;
          try {
              key = mapper.getKeyFromTuple(input);
              message = mapper.getMessageFromTuple(input);
              topic = topicSelector.getTopic(input);
              if(topic != null ) {
                  producer.send(new KeyedMessage<K, V>(topic, key, message));
              } else {
                  LOG.warn("skipping key = " + key + ", topic selector returned null.");
              }
              collector.ack(input);
          } catch (Exception ex) {
              collector.reportError(ex);
              collector.fail(input);
          }
      }
focal_method.class_method_signature: KafkaBolt.execute(Tuple input)
focal_method.full_signature: @Override public void execute(Tuple input)
focal_method.identifier: execute
focal_method.invocations: getKeyFromTuple getMessageFromTuple getTopic send warn ack reportError fail
focal_method.modifiers: @Override public
focal_method.parameters: (Tuple input)
focal_method.return: void
focal_method.signature: void execute(Tuple input)