{
    "schemaVersion": "Version of current schema", 
    "checkpointId": "Checkpoint ID", 
    "createdTimeMs": "Created time in epoch milliseconds", 
    "jobName": "Job name", 
    "jobInstance": "Instance ID of the job", 
    "taskName": "Name of the task", 
    "storeName": "Store name", 
    "dirIndex": {
        "filesPresent": [
            {
                "fileName": "Name of the file uploaded to Ambry", 
                "blobs": [
                    {
                        "blobID": "Blob ID", 
                        "offset": "File chunk offset [optional]"
                    }
                ], 
                "mtime": "Last updated time in epoch milliseconds", 
                "ctime": "Created time in epoch milliseconds",
                "sizeInBytes": "Size of the file", 
                "crc": "crc32 checksum of the file", 
                "attribs": "permission attributes of the file"
            }
        ],
        "filesRemoved": [
            {
                "fileName": "Name of the file to delete from Ambry", 
                "blobs": [
                    {
                        "blobID": "Blob ID", 
                        "offset": "File chunk offset [optional]"
                    }
                ]
            }
        ],
        "subDirectoriesPresent": [
            {
                "dirIndex": {
                     ……
                 }
            }
        ],
        "subDirectoriesRemoved": [
            {
                "dirIndex": {
                     ……
                 }
            }
        ]
    }, 
  "prevSnapshotIndexBlobId": “Blob ID of previous snapshot index”
}

/**
 * Provides interface for common blob store operations: GET, PUT and DELETE
 */
@InterfaceStability.Unstable
public interface BlobStoreManager {
  /**
   * Initialize underlying blob store client, if necessary.
   *
   */
  void init();

  /**
   * Non-blocking PUT call to remote blob store with supplied metadata
   * @param inputStream InputStream to read the file
   * @param metadata user supplied {@link Metadata} of the request
   * @return a future containing the blob ID of the uploaded blob if the upload is successful.
   */
  CompletionStage<String> put(InputStream inputStream, Metadata metadata);

  /**
   * Non-blocking GET call to remote blob store
   * @param id Blob ID of the blob to get
   * @param outputStream OutputStream to write the downloaded blob
   * @param metadata User supplied {@link Metadata} of the request
   * @return A future that completes when all the chunks are downloaded and written successfully to the OutputStream
   * @throws org.apache.samza.storage.blobstore.exceptions.DeletedException returned future should complete
   *         exceptionally with DeletedException on failure with the blob already deleted error.
   */
  CompletionStage<Void> get(String id, OutputStream outputStream, Metadata metadata);

  /**
   * Non-blocking call to mark a blob for deletion in the remote blob store
   * @param id Blob ID of the blob to delete
   * @param metadata User supplied {@link Metadata} of the request
   * @return A future that completes when the blob is successfully deleted from the blob store.
   * @throws org.apache.samza.storage.blobstore.exceptions.DeletedException returned future should complete
   *         exceptionally with DeletedException on failure with the blob already deleted error. This exception is
   *         caught and ignored by the caller of the delete method during initial cleanup and SnapshotIndex read.
   */
  CompletionStage<Void> delete(String id, Metadata metadata);

  /**
   * Non-blocking call to remove the Time-To-Live (TTL) for a blob and make it permanent.
   * @param blobId Blob ID of blob to remove TTL for.
   * @param metadata User supplied {@link Metadata} of the request
   * @return a future that completes when the TTL for the blob is removed.
   * @throws org.apache.samza.storage.blobstore.exceptions.DeletedException returned future should complete
   *         exceptionally with DeletedException on failure with the blob already deleted error.
   */
  CompletionStage<Void> removeTTL(String blobId, Metadata metadata);

  /**
   * Cleanly close resources like blob store client
   */
  void close();
}

/**
 * Factory to create instance of {@link StateBackendAdmin}s that needs to be implemented for every
 * state backend
 */
public interface BlobStoreAdminFactory {
  /**
   * Returns an instance of {@link StateBackendAdmin}
   * @param config job configuration
   * @param jobModel Job Model
   */
  StateBackendAdmin getStateBackendAdmin(Config config, JobModel jobModel);
}

/**
 * Factory to create instance of {@link BlobStoreManager}s that needs to be implemented for every state backend 
 */
public interface BlobStoreManagerFactory {
  /**
   * Returns an instance of {@link BlobStoreManager} for backup. 
   * @param config job configuration.
   * @param backupExecutor excutor service for backup operation.
   */
  BlobStoreManager getBackupBlobStoreManager(Config config, ExecutorService backupExecutor);
  
  /**
   * Returns an instance of {@link BlobStoreManager} for restore. 
   * @param config job configuration.
   * @param restoreExecutor excutor service for restore operation.
   */
  BlobStoreManager getRestoreBlobStoreManager(Config config, ExecutorService restoreExecutor);
}

/**
 * Representation of a directory in the blob store
 */
public class DirIndex {
  public static final String ROOT_DIR_NAME = "";
  private static final short SCHEMA_VERSION = 1;

  private final String dirName;

  private final List<FileIndex> filesPresent;
  private final List<FileIndex> filesRemoved;

  // Note: subDirsPresent can also have filesRemoved and subDirsRemoved within them.
  private final List<DirIndex> subDirsPresent;
  private final List<DirIndex> subDirsRemoved;
}

/**
 * Representation of a file in blob store
 */
public class FileIndex {
  private final String fileName;
  /**
   * Chunks of file uploaded to blob store as {@link FileBlob}s
   */
  private final List<FileBlob> fileBlobs;
  /**
   * Metadata (e.g. POSIX file attributes) associated with the file.
   */
  private final FileMetadata fileMetadata;
  /**
   * Checksum of the file for verifying integrity.
   */
  private final long checksum;
}

/**
 * Representation of a File in a Blob store
 */
public class FileBlob {

  private final String blobId;
  /**
   * Offset of this blob in the file. A file can be uploaded multiple chunks, and can have
   * multiple blobs associated with it. Each blob then has its own ID and an offset in the file.
   */
  private final int offset;
}

/**
 * A {@link SnapshotIndex} contains all the information necessary for recreating the local store by
 * downloading its contents from the remote blob store. The {@link SnapshotIndex} is itself serialized
 * and stored as a blob in the remote store, and its blob id tracked in the Task checkpoint.
 */
public class SnapshotIndex {
  private static final short SCHEMA_VERSION = 1;

  private final long creationTimeMillis;
  /**
   * Metadata for a snapshot like job name, job Id, store name etc.
   */
  private final SnapshotMetadata snapshotMetadata;
  private final DirIndex dirIndex;

  /**
   * Blob ID of previous snapshot index blob. Tracked here to be cleaned up
   * in cleanup phase of commit lifecycle.
   */
  private final Optional<String> prevSnapshotIndexBlobId;
}

/**
 * Representation of the diff between a local directory and a remote directory contents.
 */
public class DirDiff {

  private final String dirName;

  /**
   * New files in this directory that needs to be uploaded to the blob store.
   */
  private final List<File> filesAdded;

  /**
   * Files that have already been uploaded to the blob store in a previous snapshot and haven't changed.
   */
  private final List<FileIndex> filesRetained;

  /**
   * Files that have already been uploaded to the blob store in a previous snapshot and need to be removed.
   */
  private final List<FileIndex> filesRemoved;

  /**
   * Subdirectories of this directory that are not already present in the previous snapshot and all of their contents
   * need to be recursively added.
   */
  private final List<DirDiff> subDirsAdded;

  /**
   * Subdirectories of this directory that are already present in the previous snapshot, but whose contents
   * may have changed and may need to be recursively added or removed.
   */
  private final List<DirDiff> subDirsRetained;

  /**
   * Subdirectories that are already present in the previous snapshot, but don't exist in the local snapshot,
   * and hence all of their contents need to be recursively removed.
   */
  private final List<DirIndex> subDirsRemoved;
}