code: @Override
      public void start() throws Exception {
          executor = Executors.newCachedThreadPool(
                  new ThreadFactoryBuilder().setNameFormat("KafkaConsumer-%d").build());
          connector = Consumer.createJavaConsumerConnector(new ConsumerConfig(consumerProps));
  
          final Map<String, List<KafkaStream<byte[], byte[]>>> streams = connector.createMessageStreams(ImmutableMap.of(topic, readers));
  
          final List<KafkaStream<byte[], byte[]>> streamList = streams.get(topic);
          if (streamList == null) {
              throw new RuntimeException(topic + " is not valid");
          }
  
          running = true;
          for (KafkaStream<byte[], byte[]> stream : streamList) {
              final ConsumerIterator<byte[], byte[]> iterator = stream.iterator();
              runners.add(
                      executor.submit(new Runnable() {
                          @Override
                          public void run() {
                              while (running) {
                                  try {
                                      long pause = Math.min(pausedTime.get(), MAX_PAUSE);
                                      if (pause > 0) {
                                          Thread.sleep(pause);
                                          pausedTime.set(0);
                                      }
                                      byte[] message = iterator.next().message();
                                      router.process(
                                              KafkaConsumer.this,
                                              new DefaultMessageContainer(new Message(topic, message), jsonMapper));
                                  } catch (ConsumerTimeoutException timeoutException) {
                                      // do nothing
                                  } catch (Exception e) {
                                      log.error("Exception on consuming kafka with topic: " + topic, e);
                                  }
                              }
                          }
                      })
              );
          }
      }
docs: not final for the test