void startConsumer() {
    // set the offset for each TopicPartition
    if (topicPartitionsToOffset.size() <= 0) {
      LOG.error("{}: Consumer is not subscribed to any SSPs", this);
    }

    topicPartitionsToOffset.forEach((topicPartition, startingOffsetString) -> {
      long startingOffset = Long.valueOf(startingOffsetString);

      try {
        synchronized (kafkaConsumer) {
          kafkaConsumer.seek(topicPartition, startingOffset);
        }
      } catch (Exception e) {
        // all recoverable execptions are handled by the client.
        // if we get here there is nothing left to do but bail out.
        String msg = String.format("%s: Got Exception while seeking to %s for partition %s", this, startingOffsetString, topicPartition);
        LOG.error(msg, e);
        throw new SamzaException(msg, e);
      }

      LOG.info("{}: Changing consumer's starting offset for partition {} to {}", this, topicPartition, startingOffsetString);

      // add the partition to the proxy
      proxy.addTopicPartition(KafkaUtil.toSystemStreamPartition(systemName, topicPartition), startingOffset);
    });

    // start the proxy thread
    if (proxy != null && !proxy.isRunning()) {
      LOG.info("{}: Starting proxy {}", this, proxy);
      proxy.start();
    }
  }
Set the offsets to start from.
Register the TopicPartitions with the proxy.
Start the proxy.
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemConsumer.java