@VisibleForTesting
  void writeChangelogOffsetFile(String storeName, SystemStreamPartition ssp, String newestOffset,
      File writeDirectory) throws IOException {
    LOG.debug("Storing newest offset {} for taskName {} store {} changelog ssp {} in OFFSET file at path: {}.",
        newestOffset, taskName, storeName, ssp, writeDirectory);
    storageManagerUtil.writeOffsetFile(writeDirectory, Collections.singletonMap(ssp, newestOffset), false);
  }
Writes the newest changelog ssp offset for each logged and persistent store to the OFFSET file in the current
store directory (for allowing rollbacks). If the Kafka transactional backup manager is enabled, also writes to
the store checkpoint directory.

These files are used during container startup to ensure transactional state, and to determine whether the
there is any new information in the changelog that is not reflected in the on-disk copy of the store.
If there is any delta, it is replayed from the changelog. E.g. this can happen if the job was run on this host,
then another host, and then back to this host.
samza-core/src/main/java/org/apache/samza/storage/TaskStorageCommitManager.java