/**
 * Visitor interface for system consumers to implement to support {@link Startpoint}s.
 */
public interface StartpointVisitor {

  /**
   * Seek to specific offset represented by {@link StartpointSpecific}
   * @param systemStreamPartition The {@link SystemStreamPartition} to seek the offset to.
   * @param startpointSpecific The {@link Startpoint} that represents the specific offset.
   */
  void visit(SystemStreamPartition systemStreamPartition, StartpointSpecific startpointSpecific);

  /**
   * Seek to timestamp offset represented by {@link StartpointTimestamp}
   * @param systemStreamPartition The {@link SystemStreamPartition} to seek the offset to.
   * @param startpointTimestamp The {@link Startpoint} that represents the timestamp offset.
   */
  default void visit(SystemStreamPartition systemStreamPartition, StartpointTimestamp startpointTimestamp) {
    throw new UnsupportedOperationException("StartpointTimestamp is not supported.");
  }

  /**
   * Seek to earliest offset represented by {@link StartpointOldest}
   * @param systemStreamPartition The {@link SystemStreamPartition} to seek the offset to.
   * @param startpointOldest The {@link Startpoint} that represents the earliest offset.
   */
  default void visit(SystemStreamPartition systemStreamPartition, StartpointOldest startpointOldest) {
    throw new UnsupportedOperationException("StartpointOldest is not supported.");
  }

  /**
   * Seek to latest offset represented by {@link StartpointUpcoming}
   * @param systemStreamPartition The {@link SystemStreamPartition} to seek the offset to.
   * @param startpointUpcoming The {@link Startpoint} that represents the latest offset.
   */
  default void visit(SystemStreamPartition systemStreamPartition, StartpointUpcoming startpointUpcoming) {
    throw new UnsupportedOperationException("StartpointUpcoming is not supported.");
  }

  /**
   * Bootstrap signal represented by {@link StartpointCustom}
   * @param systemStreamPartition The {@link SystemStreamPartition} to seek the offset to.
   * @param startpointCustom The {@link Startpoint} that represents the bootstrap signal.
   */
  default void visit(SystemStreamPartition systemStreamPartition, StartpointCustom startpointCustom) {
    throw new UnsupportedOperationException(String.format("%s is not supported.", startpointCustom.getClass().getSimpleName()));
  }
}
 
// Below is example pseudocode for system specific implementations on how to handle timestamp position types. Other position types,
// (StartpointSpecific, StartpointOldest, StartpointUpcoming) left out for brevity.
 
public KafkaSystemConsumer implements SystemConsumer, StartpointVisitor {
  @Override
  public void visit(SystemStreamPartition systemStreamPartition, StartpointTimestamp startpoint) {
    if (startpoint.getPositionType() == PositionType.TIMESTAMP) {
      // Call underlying Kafka Consumer#offsetsForTimes().
      // Call underlying Kafka Consumer#seek() with offsets returned from offsetsForTimes()
    } else if...
  }
}
 
public EventHubSystemConsumer implements SystemConsumer, StartpointVisitor {
  @Override
  public void visit(SystemStreamPartition systemStreamPartition, Startpoint startpoint) {
    if (startpoint.getPositionType() == PositionType.TIMESTAMP) {
      // eventPosition = EventPosition.fromEnqueuedTime(toDateTime(startpoint.position()))
      // eventHubClientManager.getEventHubClient().createReceiverSync(consumerGroup, systemStramPartition.getPartitionId(), eventPosition)
    } else if...
  }
}

/**
 * The StartpointManager reads and writes {@link Startpoint} to the metadata store defined by
 * the configuration task.startpoint.metadata.store.factory.
 */
public class StartpointManager {
  /**
   * Writes a {@link Startpoint} that defines the start position for a {@link SystemStreamPartition}.
   *
   * @param ssp The {@link SystemStreamPartition} to map the {@link Startpoint} against.
   * @param startpoint Reference to a Startpoint object.
   */
  public void writeStartpoint(SystemStreamPartition ssp, Startpoint startpoint) {...}


  /**
   * Writes a {@link Startpoint} that defines the start position for a specified {@link SystemStreamPartition} for a specific {@link TaskName}.
   *
   * @param ssp The {@link SystemStreamPartition} to map the {@link Startpoint} against.
   * @param taskName The {@link TaskName} to map the {@link Startpoint} against.
   * @param startpoint Reference to a Startpoint object.
   */
  public void writeStartpoint(SystemStreamPartition ssp, TaskName taskName, Startpoint startpoint) {...}

/**
 * The StartpointManager reads and writes {@link Startpoint} to the metadata store defined by
 * the configuration task.startpoint.metadata.store.factory.
 */
public class StartpointManager {
  /**
   * Returns Writesthe alast {@link Startpoint} that defines the start position for a {@link SystemStreamPartition}.
   *
   * @param ssp The {@link SystemStreamPartition} to mapfetch the {@link Startpoint} againstfor.
   * @param startpoint Reference to a Startpoint object@return {@link Startpoint} for the {@link SystemStreamPartition}, or null if it does not exist.
   */
  public voidStartpoint writeStartpointreadStartpoint(SystemStreamPartition ssp, Startpoint startpoint) {...}


  /**
   * Returns Writesthe alast {@link Startpoint} that defines the start position for a specified {@link SystemStreamPartition} for a specific {@link TaskName}.
   *
   * @param ssp The {@link SystemStreamPartition} to mapfetch the {@link Startpoint} againstfor.
   * @param taskName The {@link TaskName} to map the {@link Startpoint} against.
   * @param startpoint Reference to a Startpoint object @return {@link Startpoint} for the {@link SystemStreamPartition}, or null if it does not exist.
   */
  public voidStartpoint writeStartpointForTaskreadStartpoint(SystemStreamPartition ssp, TaskName taskName, Startpoint startpoint) {...}

  /**
   * ReturnsDeletes the last {@link Startpoint} that definesfor the start position for aspecified {@link SystemStreamPartition}.
   *
   * @param ssp The {@link SystemStreamPartition} to fetch the {@link Startpoint} for.
   * @return {@link Startpoint} for the {@link SystemStreamPartition}, or null if it does not exist.
   */
  public Startpointvoid readStartpointdeleteStartpoint(SystemStreamPartition ssp) {...}

  /**
   * ReturnsDeletes the last {@link Startpoint} for athe specified {@link SystemStreamPartition} for a specific {@link TaskName}.
   *
   * @param ssp The {@link SystemStreamPartition} to fetch the {@link Startpoint} for.
   * @return@param taskName The {@link StartpointTaskName} forto map the {@link SystemStreamPartitionStartpoint}, or null if it does not exist against.
   */
  public Startpointvoid readStartpointForTaskdeleteStartpoint(SystemStreamPartition ssp, TaskName taskName) {...}

  /**
   * Deletes theFor {@link Startpoint}s forkeyed theonly specifiedby {@link SystemStreamPartition}
, this method *
re-maps the  *Startpoints @paramfrom
 ssp The {@link* SystemStreamPartition} to fetch the SystemStreamPartition+{@link StartpointTaskName} for.
 all tasks */
provided by publicthe void deleteStartpoint(SystemStreamPartition ssp) {...}

  public void deleteStartpointForTask(SystemStreamPartition ssp, TaskName taskName) {...}


  /**
   * For {@link Startpoint}s keyed only by {@link SystemStreamPartition}, this method remaps{@link JobModel}
   * This method is not atomic or thread-safe. The intent is for the Samza Processor's coordinator to use this
   * method to assign the Startpoints forto the appropriate specified
   * SystemStreamPartition to SystemStreamPartition+tasks.
   * @param jobModel The {@link JobModel} is used to determine which {@link TaskName} for all tasks provided by the each {@link SystemStreamPartitionGrouperSystemStreamPartition} maps to.
   *
 @return The list *of @implNote This method is not atomic or thread-safe. The intent is for the Samza Processor's coordinator to use this
   * method to assign the Startpoints to the appropriate tasks.
   * @param ssp The {@link SystemStreamPartition} that a {@link Startpoint} is keyed to.
   * @param grouper The {@link SystemStreamPartitionGrouper} is used to determine what the task names are.
   * @return The list of {@link TaskName}s
   */
  public Set<TaskName> groupStartpointsPerTask(SystemStreamPartition ssp, SystemStreamPartitionGrouper grouper) {...}

// Full implementation not shown for brevity
}

// OffsetManager is written in Scala, but is translated to Java here for consistency in this design document.
public class OffsetManager {
  private final CheckpointManager checkpointManager;
  private final StartpointManager startpointManager;

  // Will add similar doc as StartpointManager#readStartpoint
  public Startpoint getStartpoint(StartpointKey startpointKey) {
    return startpointManager.readStartpoint(startpointKey);
  }

  // Will add similar doc as StartpointManager#writeStartpoint
  public Startpoint setStartpoint(StartpointKey startpointKey, Startpoint startpoint) {
    startpointManager.writeStartpoint(startpointKey, startpoint);
  }
  
  // Renamed from getStartingOffset to self document this method where the offset is coming from.
  public String getStartingOffsetFromCheckpoint(TaskName taskName, SystemStreamPartition systemStreamPartition) { 
    ... 

// OffsetManager is written in Scala, but is translated to Java here for consistency in this design document.
public class OffsetManager {
  private final CheckpointManager checkpointManager;
  private final StartpointManager startpointManager;

  // Will add similar doc as StartpointManager#readStartpoint
  public Startpoint getStartpoint(StartpointKey startpointKey) {
    return startpointManager.readStartpoint(startpointKey);
  }

  // Will add similar doc as StartpointManager#writeStartpoint Alias old name for backwards-compatibility.
  @Deprecated // set as legacy and remove in next major version.
  public StartpointString setStartpointgetStartingOffset(StartpointKeyTaskName startpointKeytaskName, StartpointSystemStreamPartition startpointsystemStreamPartition) {
    startpointManager.writeStartpoint(startpointKeyreturn getStartingOffsetFromCheckpoint(taskName, startpointsystemStreamPartition);
  }
  
  //public Renamedvoid from getStartingOffset to self document this method where the offset is coming fromwriteCheckpoint(TaskName taskName, Checkpoint checkpoint) {
    ...
  public String getStartingOffsetFromCheckpointcheckpointManager.writeCheckpoint(TaskName taskName, SystemStreamPartition systemStreamPartition) { checkpoint);
    ... 
  }

  // Aliasfor oldSSPs namein for backwards-compatibility.this task
  @Deprecated // set as legacy and remove in next major version.
  public String getStartingOffset(TaskName taskName, SystemStreamPartition systemStreamPartition) {
    return getStartingOffsetFromCheckpoint(taskName, systemStreamPartition); startpointManager.deleteStartpoint(ssp, taskName);
    // 
    ...
  }

// Excluding members and methods not pertinent to this design document for brevity.
}