public Optional<Startpoint> readStartpoint(Map<String, byte[]> startpointMap, SystemStreamPartition ssp, TaskName taskName) {
    Preconditions.checkState(!stopped, "Underlying metadata store not available");
    Preconditions.checkNotNull(ssp, "SystemStreamPartition cannot be null");

    byte[] startpointBytes = startpointMap.get(toReadWriteStoreKey(ssp, taskName));

    if (ArrayUtils.isNotEmpty(startpointBytes)) {
      try {
        Startpoint startpoint = objectMapper.readValue(startpointBytes, Startpoint.class);
        if (Instant.now().minus(DEFAULT_EXPIRATION_DURATION).isBefore(Instant.ofEpochMilli(startpoint.getCreationTimestamp()))) {
          return Optional.of(startpoint); // return if deserializable and if not stale
        }
        LOG.warn("Creation timestamp: {} of startpoint: {} has crossed the expiration duration: {}. Ignoring it",
            startpoint.getCreationTimestamp(), startpoint, DEFAULT_EXPIRATION_DURATION);
      } catch (IOException ex) {
        throw new SamzaException(ex);
      }
    }

    return Optional.empty();
  }
Returns the {@link Startpoint} for a {@link SystemStreamPartition} and {@link TaskName}.
@param ssp The {@link SystemStreamPartition} to fetch the {@link Startpoint} for.
@param taskName The {@link TaskName} to fetch the {@link Startpoint} for.
@return {@link Optional} of {@link Startpoint} for the {@link SystemStreamPartition} and {@link TaskName}.
        It is empty if it does not exist or if it is too stale.
samza-core/src/main/java/org/apache/samza/startpoint/StartpointManager.java