class 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);
        }
    }

}