@Override
    public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
        LOG.info("Preparing bolt with configuration {}", this);
        //for backward compatibility.
        if (mapper == null) {
            LOG.info("Mapper not specified. Setting default mapper to {}", FieldNameBasedTupleToKafkaMapper.class.getSimpleName());
            this.mapper = new FieldNameBasedTupleToKafkaMapper<K, V>();
        }

        //for backward compatibility.
        if (topicSelector == null) {
            if (topoConf.containsKey(TOPIC)) {
                LOG.info("TopicSelector not specified. Using [{}] for topic [{}] specified in bolt configuration,",
                        DefaultTopicSelector.class.getSimpleName(), topoConf.get(TOPIC));
                this.topicSelector = new DefaultTopicSelector((String) topoConf.get(TOPIC));
            } else {
                throw new IllegalStateException("topic should be specified in bolt's configuration");
            }
        }

        if (providedCallback != null) {
            providedCallback.prepare(topoConf, context);
        }

        producer = mkProducer(boltSpecifiedProperties);
        this.collector = collector;
    }
@Test
    public void testSimple() {
        MockProducer<String, String> producer = new MockProducer<>(
                Cluster.empty(), false,
                null, new StringSerializer(), new StringSerializer());
        KafkaBolt<String, String> bolt = makeBolt(producer);

        OutputCollector collector = mock(OutputCollector.class);
        TopologyContext context = mock(TopologyContext.class);
        Map<String, Object> conf = new HashMap<>();
        bolt.prepare(conf, context, collector);

        String key = "KEY";
        String value = "VALUE";
        Tuple testTuple = createTestTuple(key, value);
        bolt.execute(testTuple);

        assertThat(producer.history().size(), is(1));
        ProducerRecord<String, String> arg = producer.history().get(0);

        LOG.info("GOT {} ->", arg);
        LOG.info("{}, {}, {}", arg.topic(), arg.key(), arg.value());
        assertThat(arg.topic(), is("MY_TOPIC"));
        assertThat(arg.key(), is(key));
        assertThat(arg.value(), is(value));

        // Complete the send
        producer.completeNext();
        verify(collector).ack(testTuple);
    }