@Override
    public String visit(SystemStreamPartition systemStreamPartition, StartpointOldest startpointOldest) {
      TopicPartition topicPartition = toTopicPartition(systemStreamPartition);
      Map<TopicPartition, Long> topicPartitionToOffsets = threadSafeKafkaConsumer.execute(consumer -> consumer.beginningOffsets(ImmutableSet.of(topicPartition)));
      Long beginningOffset = topicPartitionToOffsets.get(topicPartition);
      LOG.info("Beginning offset for topic partition: {} is {}.", topicPartition, beginningOffset);
      return String.valueOf(beginningOffset);
    }
Offers a kafka specific implementation of {@link StartpointVisitor} that resolves
different types of {@link Startpoint} to samza offset.
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java