@Override
    public String visit(SystemStreamPartition systemStreamPartition, StartpointTimestamp startpointTimestamp) {
      String streamName = systemStreamPartition.getStream();
      EventHubClientManager eventHubClientManager = eventHubSystemAdmin.getOrCreateStreamEventHubClient(streamName);
      EventHubClient eventHubClient = eventHubClientManager.getEventHubClient();

      PartitionReceiver partitionReceiver = null;
      try {
        // 1. Initialize the arguments required for creating the partition receiver.
        String partitionId = String.valueOf(systemStreamPartition.getPartition().getPartitionId());
        Instant epochInMillisInstant = Instant.ofEpochMilli(startpointTimestamp.getTimestampOffset());
        EventPosition eventPosition = EventPosition.fromEnqueuedTime(epochInMillisInstant);
        String consumerGroup = eventHubConfig.getStreamConsumerGroup(systemStreamPartition.getSystem(), streamName);

        // 2. Create a partition receiver with event position defined by the timestamp.
        partitionReceiver = eventHubClient.createReceiverSync(consumerGroup, partitionId, eventPosition);

        // 3. Read a single message from the partition receiver.
        Iterable<EventData> eventHubMessagesIterator = partitionReceiver.receiveSync(1);
        ArrayList<EventData> eventHubMessageList = Lists.newArrayList(eventHubMessagesIterator);

        // 4. Validate that a single message was fetched from the broker.
        Preconditions.checkState(eventHubMessageList.size() == 1, "Failed to read messages from EventHub system.");

        // 5. Return the offset present in the metadata of the first message.
        return eventHubMessageList.get(0).getSystemProperties().getOffset();
      } catch (EventHubException e) {
        LOG.error(String.format("Exception occurred when fetching offset for timestamp: %d from the stream: %s", startpointTimestamp.getTimestampOffset(), streamName), e);
        throw new SamzaException(e);
      } finally {
        if (partitionReceiver != null) {
          try {
            partitionReceiver.closeSync();
          } catch (EventHubException e) {
            LOG.error(String.format("Exception occurred when closing partition-receiver of the stream: %s", streamName), e);
          }
        }
      }
    }
Offers a eventhub specific implementation of {@link StartpointVisitor} that resolves
different types of {@link Startpoint} to samza offset.
samza-azure/src/main/java/org/apache/samza/system/eventhub/admin/EventHubSystemAdmin.java