public void writeFileOffsets(Map<SystemStreamPartition, String> lastProcessedOffsets) {
    storeToSSps.entrySet().stream()
        .filter(entry -> isPersistedStore(entry.getKey())) // filter out in-memory side input stores
        .forEach((entry) -> {
          String storeName = entry.getKey();
          Map<SystemStreamPartition, String> offsets = entry.getValue().stream()
            .filter(lastProcessedOffsets::containsKey)
            .collect(Collectors.toMap(Function.identity(), lastProcessedOffsets::get));

          try {
            File taskStoreDir = storageManagerUtil.getTaskStoreDir(storeBaseDir, storeName, taskName, taskMode);
            storageManagerUtil.writeOffsetFile(taskStoreDir, offsets, true);
          } catch (Exception e) {
            throw new SamzaException("Failed to write offset file for side input store: " + storeName, e);
          }
        });
  }
Writes the offset files for all side input stores one by one. There is one offset file per store.
Its contents are a JSON encoded mapping from each side input SSP to its last processed offset, and a checksum.

@param lastProcessedOffsets The offset per SSP to write
samza-core/src/main/java/org/apache/samza/storage/TaskSideInputStorageManager.java