public void restoreCheckpointFiles(File checkpointDir, File storeDir) {
    // the current task store dir should already be deleted for restore
    assert !storeDir.exists();

    try {
      Files.createDirectory(storeDir.toPath());

      for (File file : checkpointDir.listFiles()) {
        String fileName = file.getName();
        File targetFile = new File(storeDir, fileName);

        if (fileName.endsWith(SST_FILE_SUFFIX)) {
          // hardlink'ing the immutable sst-files.
          Files.createLink(targetFile.toPath(), file.toPath());
        } else {
          // true copy for all other files.
          Files.copy(file.toPath(), targetFile.toPath(), StandardCopyOption.REPLACE_EXISTING);
        }
      }
    } catch (Exception e) {
      throw new SamzaException(String.format(
          "Failed to restore store from checkpoint dir: %s to current dir: %s", checkpointDir, storeDir), e);
    }
  }
Returns the path for a storage engine to create its checkpoint based on the current checkpoint id.

@param taskStoreDir directory of the store as returned by {@link #getTaskStoreDir}
@param checkpointId current checkpoint id
@return String denoting the file path of the store with the given checkpoint id
samza-core/src/main/java/org/apache/samza/storage/StorageManagerUtil.java