diff --git a/pkg/spdk/engine.go b/pkg/spdk/engine.go
index 13927c47..5b4e0f14 100644
--- a/pkg/spdk/engine.go
+++ b/pkg/spdk/engine.go
@@ -61,7 +61,8 @@ type Engine struct {
 	Head        *api.Lvol
 	SnapshotMap map[string]*api.Lvol
 
-	IsRestoring bool
+	IsRestoring           bool
+	RestoringSnapshotName string
 
 	// UpdateCh should not be protected by the engine lock
 	UpdateCh chan interface{}
@@ -1748,16 +1749,18 @@ func (e *Engine) BackupRestore(spdkClient *spdkclient.Client, backupUrl, engineN
 	e.IsRestoring = true
 
 	// TODO: support DR volume
-	if len(e.SnapshotMap) == 0 {
-		if snapshotName == "" {
-			snapshotName = util.UUID()
-			e.log.Infof("Generating a snapshot name %s for the full restore", snapshotName)
-		}
-	} else {
-		if snapshotName == "" {
-			snapshotName = util.UUID()
-			e.log.Infof("Generating a snapshot name %s for the incremental restore", snapshotName)
-		}
+	switch {
+	case snapshotName != "":
+		e.RestoringSnapshotName = snapshotName
+		e.log.Infof("Using input snapshot name %s for the restore", e.RestoringSnapshotName)
+	case len(e.SnapshotMap) == 0:
+		e.RestoringSnapshotName = util.UUID()
+		e.log.Infof("Using new generated snapshot name %s for the full restore", e.RestoringSnapshotName)
+	case e.RestoringSnapshotName != "":
+		e.log.Infof("Using existing snapshot name %s for the incremental restore", e.RestoringSnapshotName)
+	default:
+		e.RestoringSnapshotName = util.UUID()
+		e.log.Infof("Using new generated snapshot name %s for the incremental restore because e.FinalSnapshotName is empty", e.RestoringSnapshotName)
 	}
 
 	defer func() {
@@ -1791,7 +1794,7 @@ func (e *Engine) BackupRestore(spdkClient *spdkclient.Client, backupUrl, engineN
 			err = replicaServiceCli.ReplicaBackupRestore(&client.BackupRestoreRequest{
 				BackupUrl:       backupUrl,
 				ReplicaName:     replicaName,
-				SnapshotName:    snapshotName,
+				SnapshotName:    e.RestoringSnapshotName,
 				Credential:      credential,
 				ConcurrentLimit: concurrentLimit,
 			})
diff --git a/pkg/spdk/replica.go b/pkg/spdk/replica.go
index 6350fdbf..72344b4e 100644
--- a/pkg/spdk/replica.go
+++ b/pkg/spdk/replica.go
@@ -1926,11 +1926,11 @@ func (r *Replica) BackupRestore(spdkClient *spdkclient.Client, backupUrl, snapsh
 		lvolName := GetReplicaSnapshotLvolName(r.Name, snapshotName)
 		r.restore, err = NewRestore(spdkClient, lvolName, snapshotName, backupUrl, backupName, r)
 		if err != nil {
-			err = errors.Wrapf(err, "failed to start new restore")
+			err = errors.Wrap(err, "failed to start new restore")
 			return grpcstatus.Errorf(grpccodes.Internal, err.Error())
 		}
 	} else {
-		r.log.Info("Resetting the restore for backup %v", backupUrl)
+		r.log.Infof("Resetting the restore for backup %v", backupUrl)
 
 		var lvolName string
 		var snapshotNameToBeRestored string
@@ -1938,13 +1938,13 @@ func (r *Replica) BackupRestore(spdkClient *spdkclient.Client, backupUrl, snapsh
 		validLastRestoredBackup := r.canDoIncrementalRestore(restore, backupUrl, backupName)
 		if validLastRestoredBackup {
 			r.log.Infof("Starting an incremental restore for backup %v", backupUrl)
-			lvolName = GetReplicaSnapshotLvolName(r.Name, restore.LastRestored)
-			snapshotNameToBeRestored = restore.LastRestored
 		} else {
 			r.log.Infof("Starting a full restore for backup %v", backupUrl)
-			lvolName = GetReplicaSnapshotLvolName(r.Name, snapshotName)
-			snapshotNameToBeRestored = snapshotName
 		}
+
+		lvolName = GetReplicaSnapshotLvolName(r.Name, snapshotName)
+		snapshotNameToBeRestored = snapshotName
+
 		r.restore.StartNewRestore(backupUrl, backupName, lvolName, snapshotNameToBeRestored, validLastRestoredBackup)
 	}
 
@@ -2052,7 +2052,7 @@ func (r *Replica) completeBackupRestore(spdkClient *spdkclient.Client, isFullRes
 		return r.postFullRestoreOperations(spdkClient, restore)
 	}
 
-	return r.postIncrementalRestoreOperations(restore)
+	return r.postIncrementalRestoreOperations(spdkClient, restore)
 }
 
 func (r *Replica) waitForRestoreComplete() error {
@@ -2083,7 +2083,33 @@ func (r *Replica) waitForRestoreComplete() error {
 	return nil
 }
 
-func (r *Replica) postIncrementalRestoreOperations(restore *Restore) error {
+func (r *Replica) postIncrementalRestoreOperations(spdkClient *spdkclient.Client, restore *Restore) error {
+	r.log.Infof("Replacing snapshot %v of the restored volume", restore.SnapshotName)
+
+	if r.restore.State == btypes.ProgressStateCanceled {
+		r.log.Info("Doing nothing for canceled backup restoration")
+		return nil
+	}
+
+	// Delete snapshot; SPDK will coalesce the content into the current head lvol.
+	r.log.Infof("Deleting snapshot %v for snapshot replacement of the restored volume", restore.SnapshotName)
+	_, err := r.SnapshotDelete(spdkClient, restore.SnapshotName)
+	if err != nil {
+		r.log.WithError(err).Error("Failed to delete snapshot of the restored volume")
+		return errors.Wrapf(err, "failed to delete snapshot of the restored volume")
+	}
+
+	r.log.Infof("Creating snapshot %v for snapshot replacement of the restored volume", restore.SnapshotName)
+	opts := &api.SnapshotOptions{
+		UserCreated: false,
+		Timestamp:   util.Now(),
+	}
+	_, err = r.SnapshotCreate(spdkClient, restore.SnapshotName, opts)
+	if err != nil {
+		r.log.WithError(err).Error("Failed to take snapshot of the restored volume")
+		return errors.Wrapf(err, "failed to take snapshot of the restored volume")
+	}
+
 	r.log.Infof("Done running incremental restore %v to lvol %v", restore.BackupURL, restore.LvolName)
 	return nil
 }