public void stop(long timeoutMs) {
    LOG.info("Shutting down KafkaConsumerProxy poll thread {} for {}", consumerPollThread.getName(), this);

    isRunning = false;
    try {
      consumerPollThread.join(timeoutMs / 2);
      // join() may timeout
      // in this case we should interrupt it and wait again
      if (consumerPollThread.isAlive()) {
        consumerPollThread.interrupt();
        consumerPollThread.join(timeoutMs / 2);
      }
    } catch (InterruptedException e) {
      LOG.warn("Join in KafkaConsumerProxy has failed", e);
      consumerPollThread.interrupt();
    }
  }
Stop this KafkaConsumerProxy and wait for at most {@code timeoutMs}.
@param timeoutMs maximum time to wait to stop this KafkaConsumerProxy
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaConsumerProxy.java