public void stop(Map<SystemStreamPartition, String> lastProcessedOffsets) {
    LOG.info("Stopping the side input stores.");
    stores.values().forEach(StorageEngine::stop);
    writeFileOffsets(lastProcessedOffsets);
  }
Stops the storage engines for all the stores and writes the offset file to disk.

@param lastProcessedOffsets The last processed offsets for each SSP. These will be used when writing offsets files
                            for each store.
samza-core/src/main/java/org/apache/samza/storage/TaskSideInputStorageManager.java