private Runnable createProxyThreadRunnable() {
    return () -> {
      isRunning = true;

      try {
        consumerPollThreadStartLatch.countDown();
        LOG.info("Starting consumer poll thread {} for system {}", consumerPollThread.getName(), systemName);
        initializeLags();
        while (isRunning) {
          fetchMessages();
        }
      } catch (Throwable throwable) {
        LOG.error(String.format("Error in KafkaConsumerProxy poll thread for system: %s.", systemName), throwable);
        // KafkaSystemConsumer uses the failureCause to propagate the throwable to the container
        failureCause = throwable;
        isRunning = false;
        kafkaSystemConsumer.setFailureCause(this.failureCause);
      }

      if (!isRunning) {
        LOG.info("KafkaConsumerProxy for system {} has stopped.", systemName);
      }
    };
  }
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