@VisibleForTesting
  void writeChangelogOffsetFiles(Map<SystemStreamPartition, String> checkpointOffsets) {
    if (storageEngines == null) {
      throw new SamzaException(String.format(
          "Storage engines are not initialized and writeChangelogOffsetFiles not be written for task %s", taskName));
    }
    storeChangelogs.forEach((storeName, systemStream) -> {
      SystemStreamPartition changelogSSP = new SystemStreamPartition(
          systemStream.getSystem(), systemStream.getStream(), taskChangelogPartition);

      // Only write if the store is durable and persisted to disk
      if (checkpointOffsets.containsKey(changelogSSP) &&
          storageEngines.containsKey(storeName) &&
          storageEngines.get(storeName).getStoreProperties().isDurableStore() &&
          storageEngines.get(storeName).getStoreProperties().isPersistedToDisk()) {
        LOG.debug("Writing changelog offset for taskName {} store {} changelog {}.", taskName, storeName, systemStream);
        File currentStoreDir = storageManagerUtil.getTaskStoreDir(durableStoreBaseDir, storeName, taskName, TaskMode.Active);
        try {
          KafkaChangelogSSPOffset kafkaChangelogSSPOffset = KafkaChangelogSSPOffset
              .fromString(checkpointOffsets.get(changelogSSP));
          // Write offsets to file system if it is non-null
          String newestOffset = kafkaChangelogSSPOffset.getChangelogOffset();
          if (newestOffset != null) {
            // Write changelog SSP offset to the OFFSET files in the task store directory
            writeChangelogOffsetFile(storeName, changelogSSP, newestOffset, currentStoreDir);

            // Write changelog SSP offset to the OFFSET files in the store checkpoint directory
            File checkpointDir = Paths.get(storageManagerUtil.getStoreCheckpointDir(
                currentStoreDir, kafkaChangelogSSPOffset.getCheckpointId())).toFile();
            writeChangelogOffsetFile(storeName, changelogSSP, newestOffset, checkpointDir);
          } else {
            // If newestOffset is null, then it means the changelog ssp is (or has become) empty. This could be
            // either because the changelog topic was newly added, repartitioned, or manually deleted and recreated.
            // No need to persist the offset file.
            LOG.debug("Deleting OFFSET file for taskName {} store {} changelog ssp {} since the newestOffset is null.",
                taskName, storeName, changelogSSP);
            storageManagerUtil.deleteOffsetFile(currentStoreDir);
          }
        } catch (IOException e) {
          throw new SamzaException(
              String.format("Error storing offset for taskName %s store %s changelog %s.", taskName, storeName,
                  systemStream), e);
        }
      }
    });
    LOG.debug("Done writing OFFSET files for logged persistent key value stores for task {}", taskName);
  }
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