@Override
  public void start() {
    if (!started.compareAndSet(false, true)) {
      LOG.warn("{}: Attempting to start the consumer for the second (or more) time.", this);
      return;
    }
    if (stopped.get()) {
      LOG.error("{}: Attempting to start a stopped consumer", this);
      return;
    }
    // initialize the subscriptions for all the registered TopicPartitions
    startSubscription();
    // needs to be called after all the registrations are completed
    setFetchThresholds();

    startConsumer();
    LOG.info("{}: Consumer started", this);
  }
Create internal kafka consumer object, which will be used in the Proxy.
@param <K> key type for the consumer
@param <V> value type for the consumer
@param systemName system name for which we create the consumer
@param kafkaConsumerConfig config object for Kafka's KafkaConsumer
@return KafkaConsumer newly created kafka consumer object
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemConsumer.java