class SnapshotUtils {

	public static <OUT, OP extends StreamOperator<OUT>> TaggedOperatorSubtaskState snapshot(
		OP operator,
		int index,
		long timestamp,
		boolean isExactlyOnceMode,
		boolean isUnalignedCheckpoint,
		CheckpointStorageWorkerView checkpointStorage,
		Path savepointPath) throws Exception {

		CheckpointOptions options = new CheckpointOptions(
			CheckpointType.SAVEPOINT,
			AbstractFsCheckpointStorage.encodePathAsReference(savepointPath),
			isExactlyOnceMode,
			isUnalignedCheckpoint);

		operator.prepareSnapshotPreBarrier(CHECKPOINT_ID);

		CheckpointStreamFactory storage = checkpointStorage.resolveCheckpointStorageLocation(
			CHECKPOINT_ID,
			options.getTargetLocation());

		OperatorSnapshotFutures snapshotInProgress = operator.snapshotState(
			CHECKPOINT_ID,
			timestamp,
			options,
			storage);

		OperatorSubtaskState state = new OperatorSnapshotFinalizer(snapshotInProgress).getJobManagerOwnedState();

		operator.notifyCheckpointComplete(CHECKPOINT_ID);
		return new TaggedOperatorSubtaskState(index, state);
	}

	private  SnapshotUtils();

}