code: 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));
          }
      }
docs: 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.
  
  @param  topic    The topic to start streaming messages from.
  @param  streams  The streams for those messages.