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