public void init() {
    this.taskSideInputStorageManager.init();

    Map<SystemStreamPartition, String> fileOffsets = this.taskSideInputStorageManager.getFileOffsets();
    LOG.info("File offsets for the task {}: {}", taskName, fileOffsets);

    this.lastProcessedOffsets.putAll(fileOffsets);
    LOG.info("Last processed offsets for the task {}: {}", taskName, lastProcessedOffsets);

    this.startingOffsets = getStartingOffsets(fileOffsets, getOldestOffsets());
    LOG.info("Starting offsets for the task {}: {}", taskName, startingOffsets);

    this.initialSideInputSSPMetadata = getInitialSideInputSSPMetadata();
    LOG.info("Task {} will catch up to offsets {}", this.taskName, this.initialSideInputSSPMetadata);

    this.startingOffsets.forEach((ssp, offset) -> checkCaughtUp(ssp, offset, SystemStreamMetadata.OffsetType.UPCOMING));
  }
Initializes the underlying {@link TaskSideInputStorageManager} and determines starting offsets for each SSP.
samza-core/src/main/java/org/apache/samza/storage/TaskSideInputHandler.java