code: public void initializeTopic(String topic) {
          if (_topics.get(topic) == null) {
              synchronized (this) {
                  if (_topics.get(topic) == null) {
                      _logger.info("Initializing streams for topic: {}", topic);
  
                      Properties props = new Properties();
  
                      props.setProperty("zookeeper.connect",
                          _configuration.getValue(Property.ZOOKEEPER_CONNECT.getName(), Property.ZOOKEEPER_CONNECT.getDefaultValue()));
                      props.setProperty("group.id",
                          _configuration.getValue(Property.KAFKA_CONSUMER_GROUPID.getName(), Property.KAFKA_CONSUMER_GROUPID.getDefaultValue()));
                      props.setProperty("auto.offset.reset", _configuration.getValue(Property.KAFKA_CONSUMER_OFFSET_RESET.getName(), Property.KAFKA_CONSUMER_OFFSET_RESET.getDefaultValue()));
                      props.setProperty("auto.commit.interval.ms", "60000");
                      props.setProperty("fetch.message.max.bytes", "2000000");
  
                      ConsumerConnector consumer = kafka.consumer.Consumer.createJavaConsumerConnector(new ConsumerConfig(props));
                      List<KafkaStream<byte[], byte[]>> streams = _createStreams(consumer, topic);
                      Topic t = new Topic(topic, consumer, streams.size());
  
                      _topics.put(topic, t);
                      _startStreamingMessages(topic, streams);
                  }
              }
          }
      }
docs: This method creates Kafka streams for a topic so that messages can be streamed to the local buffer. If the streams for the given topic have
  already been initialized the returns. Information about a particular topic is stored in a HashMap. This method uses double-checked locking to
  make sure only one client thread can initialize streams for a topic. Moreover, it also helps subsequent calls, to check if the topic has been
  initialized, be not synchronized and hence return faster.
  
  @param  topic  The topic to initialize.