private void startSubscription() {
    //subscribe to all the registered TopicPartitions
    LOG.info("{}: Consumer subscribes to {}", this, topicPartitionsToOffset.keySet());
    try {
      synchronized (kafkaConsumer) {
        // we are using assign (and not subscribe), so we need to specify both topic and partition
        kafkaConsumer.assign(topicPartitionsToOffset.keySet());
      }
    } catch (Exception e) {
      throw new SamzaException("Consumer subscription failed for " + this, e);
    }
  }
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