KafkaBolt extends BaseRichBolt { @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); } }  KafkaBolt<K,V> withTupleToKafkaMapper(TupleToKafkaMapper<K,V> mapper); KafkaBolt<K,V> withTopicSelector(KafkaTopicSelector selector); @Override void prepare(Map stormConf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple input); @Override void declareOutputFields(OutputFieldsDeclarer declarer); static final String TOPIC; static final String KAFKA_BROKER_PROPERTIES; }
@Test public void executeWithKey() throws Exception { String message = "value-123"; String key = "key-123"; Tuple tuple = generateTestTuple(key, message); bolt.execute(tuple); verify(collector).ack(tuple); verifyMessage(key, message); }
@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 public void executeWithoutKey() throws Exception { String message = "value-234"; Tuple tuple = generateTestTuple(message); bolt.execute(tuple); verify(collector).ack(tuple); verifyMessage(null, message); }
@Test public void executeWithBrokerDown() throws Exception { broker.shutdown(); String message = "value-234"; Tuple tuple = generateTestTuple(message); bolt.execute(tuple); verify(collector).fail(tuple); }