private void _startStreamingMessages ( String topic , List < KafkaStream < byte [ ] , byte [ ] > > streams ) { ExecutorService executorService = _topics . get ( topic ) . getStreamExecutorService ( ) ; for ( final KafkaStream < byte [ ] , byte [ ] > stream : streams ) { executorService . submit ( new KafkaConsumer ( stream ) ) ; } }
Retrieves the executor service for the given topic from the map of topics and submits a KafkaConsumer task for each stream in the list of streams .