diff --git a/pkg/apis/velero/v1/labels_annotations.go b/pkg/apis/velero/v1/labels_annotations.go
index 0b105e106..15bd57d78 100644
--- a/pkg/apis/velero/v1/labels_annotations.go
+++ b/pkg/apis/velero/v1/labels_annotations.go
@@ -75,4 +75,7 @@ const (
 	// ResourceTimeoutAnnotation is the annotation key used to carry the global resource
 	// timeout value for backup to plugins.
 	ResourceTimeoutAnnotation = "velero.io/resource-timeout"
+
+	// AsyncOperationIDLabel is the label key used to identify the async operation ID
+	AsyncOperationIDLabel = "velero.io/async-operation-id"
 )
diff --git a/pkg/backup/backup_pv_action.go b/pkg/backup/backup_pv_action.go
index 56c4fff8a..56ff1f4b9 100644
--- a/pkg/backup/backup_pv_action.go
+++ b/pkg/backup/backup_pv_action.go
@@ -17,6 +17,8 @@ limitations under the License.
 package backup
 
 import (
+	"strings"
+
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	corev1api "k8s.io/api/core/v1"
@@ -70,6 +72,15 @@ func (a *PVCAction) Execute(item runtime.Unstructured, backup *v1.Backup) (runti
 		pvc.Spec.DataSourceRef = nil
 	}
 
+	// remove label selectors with "velero.io/" prefixing in the key which is left by Velero restore
+	if pvc.Spec.Selector != nil && pvc.Spec.Selector.MatchLabels != nil {
+		for k := range pvc.Spec.Selector.MatchLabels {
+			if strings.HasPrefix(k, "velero.io/") {
+				delete(pvc.Spec.Selector.MatchLabels, k)
+			}
+		}
+	}
+
 	pvcMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&pvc)
 	if err != nil {
 		return nil, nil, errors.Wrap(err, "unable to convert pvc to unstructured item")
diff --git a/pkg/cmd/server/server.go b/pkg/cmd/server/server.go
index 66f8fbdd3..d2c596f29 100644
--- a/pkg/cmd/server/server.go
+++ b/pkg/cmd/server/server.go
@@ -102,7 +102,7 @@ const (
 	defaultBackupTTL = 30 * 24 * time.Hour
 
 	defaultCSISnapshotTimeout   = 10 * time.Minute
-	defaultItemOperationTimeout = 60 * time.Minute
+	defaultItemOperationTimeout = 4 * time.Hour
 
 	resourceTimeout = 10 * time.Minute
 
@@ -229,7 +229,7 @@ func NewCommand(f client.Factory) *cobra.Command {
 	command.Flags().DurationVar(&config.itemOperationSyncFrequency, "item-operation-sync-frequency", config.itemOperationSyncFrequency, "How often to check status on backup/restore operations after backup/restore processing. Default is 10 seconds")
 	command.Flags().BoolVar(&config.defaultVolumesToFsBackup, "default-volumes-to-fs-backup", config.defaultVolumesToFsBackup, "Backup all volumes with pod volume file system backup by default.")
 	command.Flags().StringVar(&config.uploaderType, "uploader-type", config.uploaderType, "Type of uploader to handle the transfer of data of pod volumes")
-	command.Flags().DurationVar(&config.defaultItemOperationTimeout, "default-item-operation-timeout", config.defaultItemOperationTimeout, "How long to wait on asynchronous BackupItemActions and RestoreItemActions to complete before timing out. Default is 1 hour")
+	command.Flags().DurationVar(&config.defaultItemOperationTimeout, "default-item-operation-timeout", config.defaultItemOperationTimeout, "How long to wait on asynchronous BackupItemActions and RestoreItemActions to complete before timing out. Default is 4 hours")
 	command.Flags().DurationVar(&config.resourceTimeout, "resource-timeout", config.resourceTimeout, "How long to wait for resource processes which are not covered by other specific timeout parameters. Default is 10 minutes.")
 	command.Flags().IntVar(&config.maxConcurrentK8SConnections, "max-concurrent-k8s-connections", config.maxConcurrentK8SConnections, "Max concurrent connections number that Velero can create with kube-apiserver. Default is 30.")
 
diff --git a/pkg/controller/data_upload_controller.go b/pkg/controller/data_upload_controller.go
index af7c9dbdc..759b297a1 100644
--- a/pkg/controller/data_upload_controller.go
+++ b/pkg/controller/data_upload_controller.go
@@ -43,6 +43,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
+	"github.com/vmware-tanzu/velero/pkg/datamover"
 	"github.com/vmware-tanzu/velero/pkg/datapath"
 	"github.com/vmware-tanzu/velero/pkg/exposer"
 	"github.com/vmware-tanzu/velero/pkg/repository"
@@ -222,12 +223,16 @@ func (r *DataUploadReconciler) runCancelableDataUpload(ctx context.Context, du *
 	}
 
 	log.WithField("path", path.ByPath).Debug("Found host path")
-	if err := fsBackup.Init(ctx, du.Spec.BackupStorageLocation, du.Namespace, exposer.GetUploaderType(du.Spec.DataMover),
+	if err := fsBackup.Init(ctx, du.Spec.BackupStorageLocation, du.Spec.SourceNamespace, datamover.GetUploaderType(du.Spec.DataMover),
 		velerov1api.BackupRepositoryTypeKopia, "", r.repoEnsurer, r.credentialGetter); err != nil {
 		return r.errorOut(ctx, du, err, "error to initialize data path", log)
 	}
 	log.WithField("path", path.ByPath).Info("fs init")
-	if err := fsBackup.StartBackup(path, fmt.Sprintf("%s/%s", du.Spec.SourceNamespace, du.Spec.SourcePVC), "", false, nil); err != nil {
+
+	tags := map[string]string{
+		velerov1api.AsyncOperationIDLabel: du.Labels[velerov1api.AsyncOperationIDLabel],
+	}
+	if err := fsBackup.StartBackup(path, fmt.Sprintf("%s/%s", du.Spec.SourceNamespace, du.Spec.SourcePVC), "", false, tags); err != nil {
 		return r.errorOut(ctx, du, err, "error starting data path backup", log)
 	}
 
@@ -469,7 +474,6 @@ func (r *DataUploadReconciler) acceptDataUpload(ctx context.Context, du *velerov
 
 	r.logger.Infof("Accepting snapshot backup %s", du.Name)
 
-	time.Sleep(2 * time.Second)
 	// For all data upload controller in each node-agent will try to update dataupload CR, and only one controller will success,
 	// and the success one could handle later logic
 	err := r.client.Update(ctx, updated)
diff --git a/pkg/datamover/util.go b/pkg/datamover/util.go
new file mode 100644
index 000000000..757deb0d2
--- /dev/null
+++ b/pkg/datamover/util.go
@@ -0,0 +1,25 @@
+/*
+Copyright The Velero Contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package datamover
+
+func GetUploaderType(dataMover string) string {
+	if dataMover == "" || dataMover == "velero" {
+		return "kopia"
+	} else {
+		return dataMover
+	}
+}
diff --git a/pkg/exposer/types.go b/pkg/exposer/types.go
index 670b03828..91fe0d066 100644
--- a/pkg/exposer/types.go
+++ b/pkg/exposer/types.go
@@ -35,11 +35,3 @@ type ExposeByPod struct {
 	HostingPod *corev1.Pod
 	PVC        string
 }
-
-func GetUploaderType(dataMover string) string {
-	if dataMover == "" || dataMover == "velero" {
-		return "kopia"
-	} else {
-		return dataMover
-	}
-}
diff --git a/pkg/generated/clientset/versioned/typed/velero/v1/podvolumebackup.go b/pkg/generated/clientset/versioned/typed/velero/v1/podvolumebackup.go
index 2acacf8c6..836d78b58 100644
--- a/pkg/generated/clientset/versioned/typed/velero/v1/podvolumebackup.go
+++ b/pkg/generated/clientset/versioned/typed/velero/v1/podvolumebackup.go
@@ -36,7 +36,6 @@ type PodVolumeBackupsGetter interface {
 	PodVolumeBackups(namespace string) PodVolumeBackupInterface
 }
 
-//go:generate mockery --name PodVolumeBackupInterface
 // PodVolumeBackupInterface has methods to work with PodVolumeBackup resources.
 type PodVolumeBackupInterface interface {
 	Create(ctx context.Context, podVolumeBackup *v1.PodVolumeBackup, opts metav1.CreateOptions) (*v1.PodVolumeBackup, error)