@Override
    public String visit(SystemStreamPartition systemStreamPartition, StartpointTimestamp startpointTimestamp) {
      Preconditions.checkNotNull(startpointTimestamp, "Startpoint cannot be null");
      Preconditions.checkNotNull(startpointTimestamp.getTimestampOffset(), "Timestamp field in startpoint cannot be null");
      TopicPartition topicPartition = toTopicPartition(systemStreamPartition);

      Map<TopicPartition, Long> topicPartitionToTimestamp = ImmutableMap.of(topicPartition, startpointTimestamp.getTimestampOffset());
      LOG.info("Finding offset for timestamp: {} in topic partition: {}.", startpointTimestamp.getTimestampOffset(), topicPartition);
      Map<TopicPartition, OffsetAndTimestamp> topicPartitionToOffsetTimestamps = threadSafeKafkaConsumer.execute(consumer -> consumer.offsetsForTimes(topicPartitionToTimestamp));

      OffsetAndTimestamp offsetAndTimestamp = topicPartitionToOffsetTimestamps.get(topicPartition);
      if (offsetAndTimestamp != null) {
        return String.valueOf(offsetAndTimestamp.offset());
      } else {
        LOG.info("Offset for timestamp: {} does not exist for partition: {}. Falling back to end offset.", startpointTimestamp.getTimestampOffset(), topicPartition);
        return getEndOffset(systemStreamPartition);
      }
    }
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