public void start() {
    if (!consumerPollThread.isAlive()) {
      LOG.info("Starting KafkaConsumerProxy polling thread for " + this.toString());

      consumerPollThread.start();

      // we need to wait until the thread starts
      while (!isRunning && failureCause == null) {
        try {
          consumerPollThreadStartLatch.await(3000, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
          LOG.info("Ignoring InterruptedException while waiting for consumer poll thread to start.", e);
        }
      }
    } else {
      LOG.warn("Tried to start an already started KafkaConsumerProxy (%s). Ignoring.", this.toString());
    }

    if (topicPartitionToSSP.size() == 0) {
      String msg = String.format("Started KafkaConsumerProxy without any registered TopicPartitions for %s", systemName);
      LOG.warn(msg);
    }
  }
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