def initialize(self, conf, ctx):
        """Initialization steps:

        1. Connect to Kafka.
        2. Prepare Kafka producer for `tweet` topic.
        3. Prepare to track tweets published to topic, to avoid redundant data.
        """
        config = get_config()['ResultTopicBolt']
        kafka_class = import_name(config['kafka_class'])
        self.client = kafka_class(**config['kafka_init'])
        self.topic = self.client.topics[config['topic']]
        self.producer = self.topic.get_producer()

        # Use own default index value while still allowing user config.
        self.tweet_shelf = shelf_from_config(config, index='pre_kafka_shelf')