public void writeOffsetFile(File storeDir, Map<SystemStreamPartition, String> offsets, boolean isSideInput) throws IOException {

    // First, we write the new-format offset file
    File offsetFile = new File(storeDir, OFFSET_FILE_NAME_NEW);
    String fileContents = SSP_OFFSET_OBJECT_WRITER.writeValueAsString(offsets);
    FileUtil fileUtil = new FileUtil();
    fileUtil.writeWithChecksum(offsetFile, fileContents);

    // Now we write the old format offset file, which are different for store-offset and side-inputs
    if (isSideInput) {
      SIDE_INPUT_OFFSET_FILE_SEMAPHORE.acquireUninterruptibly();
      try {
        offsetFile = new File(storeDir, SIDE_INPUT_OFFSET_FILE_NAME_LEGACY);
        fileContents = SSP_OFFSET_OBJECT_WRITER.writeValueAsString(offsets);
        fileUtil.writeWithChecksum(offsetFile, fileContents);
      } finally {
        SIDE_INPUT_OFFSET_FILE_SEMAPHORE.release();
      }
    } else {
      offsetFile = new File(storeDir, OFFSET_FILE_NAME_LEGACY);
      fileUtil.writeWithChecksum(offsetFile, offsets.entrySet().iterator().next().getValue());
    }
  }
Write the given SSP-Offset map into the offsets file.
@param storeDir the directory of the store
@param offsets The SSP-offset to write
@param isSideInput true if store is a side-input store, false if it is a regular store
@throws IOException because of deserializing to json
samza-core/src/main/java/org/apache/samza/storage/StorageManagerUtil.java