public void copySideInputOffsetFileToCheckpointDir(File storeBaseDir,
      TaskName taskName, String storeName, TaskMode taskMode, CheckpointId checkpointId) {
    File storeDir = getTaskStoreDir(storeBaseDir, storeName, taskName, taskMode);
    File checkpointDir = new File(getStoreCheckpointDir(storeDir, checkpointId));

    SIDE_INPUT_OFFSET_FILE_SEMAPHORE.acquireUninterruptibly();
    try {
      File storeSideInputOffsetsFile = new File(storeDir, SIDE_INPUT_OFFSET_FILE_NAME_LEGACY);
      File checkpointDirSideInputOffsetsFile = new File(checkpointDir, SIDE_INPUT_OFFSET_FILE_NAME_LEGACY);
      if (storeSideInputOffsetsFile.exists()) {
        FileUtils.copyFile(storeSideInputOffsetsFile, checkpointDirSideInputOffsetsFile);
      } else {
        LOG.info("Did not find the file to copy: {} for taskName: {} taskMode: {} storeName: {} checkpointId: {}",
            SIDE_INPUT_OFFSET_FILE_NAME_LEGACY, taskName, taskMode, storeName, checkpointId);
      }
    } catch (IOException e) {
      String msg = String.format(
          "Error copying %s file to checkpoint dir for taskName: %s taskMode: %s storeName: %s checkpointId: %s",
          SIDE_INPUT_OFFSET_FILE_NAME_LEGACY, taskName, taskMode, storeName, checkpointId);
      LOG.error(msg, e);
      throw new SamzaException(msg, e);
    } finally {
      SIDE_INPUT_OFFSET_FILE_SEMAPHORE.release();
    }
  }
Writes the checkpoint to the store checkpoint directory based on the checkpointId.

@param storeDir store or store checkpoint directory to write the checkpoint to
@param checkpoint checkpoint v2 containing the checkpoint Id
samza-core/src/main/java/org/apache/samza/storage/StorageManagerUtil.java