diff --git a/internal/controller/vrg_volrep.go b/internal/controller/vrg_volrep.go
index 5904ae767..4ca9d1feb 100644
--- a/internal/controller/vrg_volrep.go
+++ b/internal/controller/vrg_volrep.go
@@ -32,6 +32,20 @@ const (
 	defaultVRCAnnotationKey = "replication.storage.openshift.io/is-default-class"
 )
 
+const (
+	// secretRef keys
+	controllerPublishSecretName = "csi.storage.k8s.io/controller-publish-secret-name"
+	controllerPublishSecretNamespace = "csi.storage.k8s.io/controller-publish-secret-namespace"
+	nodeStageSecretName             = "csi.storage.k8s.io/node-stage-secret-name"
+	nodeStageSecretNamespace        = "csi.storage.k8s.io/node-stage-secret-namespace"
+	nodePublishSecretName           = "csi.storage.k8s.io/node-publish-secret-name"
+	nodePublishSecretNamespace      = "csi.storage.k8s.io/node-publish-secret-namespace"
+	controllerExpandSecretName      = "csi.storage.k8s.io/controller-expand-secret-name"
+	controllerExpandSecretNamespace = "csi.storage.k8s.io/controller-expand-secret-namespace"
+	nodeExpandSecretName            = "csi.storage.k8s.io/node-expand-secret-name"
+	nodeExpandSecretNamespace       = "csi.storage.k8s.io/node-expand-secret-namespace"
+)
+
 func logWithPvcName(log logr.Logger, pvc *corev1.PersistentVolumeClaim) logr.Logger {
 	return log.WithValues("pvc", types.NamespacedName{Name: pvc.Name, Namespace: pvc.Namespace}.String())
 }
@@ -1332,6 +1346,24 @@ func (v *VRGInstance) filterDefaultVRC(
 		defaultVRCAnnotationKey)
 }
 
+func (v *VRGInstance) getStorageClassFromSCName(scName *string) (*storagev1.StorageClass, error) {
+	if storageClass, ok := v.storageClassCache[*scName]; ok {
+		return storageClass, nil
+	}
+
+	storageClass := &storagev1.StorageClass{}
+	if err := v.reconciler.Get(v.ctx, types.NamespacedName{Name: *scName}, storageClass); err != nil {
+		v.log.Info(fmt.Sprintf("Failed to get the storageclass %s", *scName))
+
+		return nil, fmt.Errorf("failed to get the storageclass with name %s (%w)",
+			*scName, err)
+	}
+
+	v.storageClassCache[*scName] = storageClass
+
+	return storageClass, nil
+}
+
 // getStorageClass inspects the PVCs being protected by this VRG instance for the passed in namespacedName, and
 // returns its corresponding StorageClass resource from an instance cache if available, or fetches it from the API
 // server and stores it in an instance cache before returning the StorageClass
@@ -1363,21 +1395,7 @@ func (v *VRGInstance) getStorageClass(namespacedName types.NamespacedName) (*sto
 		return nil, fmt.Errorf("missing StorageClass name for pvc (%s)", namespacedName)
 	}
 
-	if storageClass, ok := v.storageClassCache[*scName]; ok {
-		return storageClass, nil
-	}
-
-	storageClass := &storagev1.StorageClass{}
-	if err := v.reconciler.Get(v.ctx, types.NamespacedName{Name: *scName}, storageClass); err != nil {
-		v.log.Info(fmt.Sprintf("Failed to get the storageclass %s", *scName))
-
-		return nil, fmt.Errorf("failed to get the storageclass with name %s (%w)",
-			*scName, err)
-	}
-
-	v.storageClassCache[*scName] = storageClass
-
-	return storageClass, nil
+	return v.getStorageClassFromSCName(scName)
 }
 
 // checkVRStatus checks if the VolumeReplication resource has the desired status for the
@@ -2025,7 +2043,7 @@ func (v *VRGInstance) restorePVsFromObjectStore(objectStore ObjectStorer, s3Prof
 		return 0, fmt.Errorf("%s: %w", errMsg, err)
 	}
 
-	return restoreClusterDataObjects(v, pvList, "PV", cleanupPVForRestore, v.validateExistingPV)
+	return restoreClusterDataObjects(v, pvList, "PV", v.cleanupPVForRestore, v.validateExistingPV)
 }
 
 func (v *VRGInstance) restorePVCsFromObjectStore(objectStore ObjectStorer, s3ProfileName string) (int, error) {
@@ -2092,7 +2110,7 @@ func restoreClusterDataObjects[
 ](
 	v *VRGInstance,
 	objList []ObjectType, objType string,
-	cleanupForRestore func(*ObjectType),
+	cleanupForRestore func(*ObjectType) error,
 	validateExistingObject func(*ObjectType) error,
 ) (int, error) {
 	numRestored := 0
@@ -2102,7 +2120,13 @@ func restoreClusterDataObjects[
 		objectCopy := &*object
 		obj := ClientObject(objectCopy)
 
-		cleanupForRestore(objectCopy)
+		err := cleanupForRestore(objectCopy)
+		if err != nil {
+			v.log.Info("failed to cleanup during restore", "error", err.Error())
+
+			return numRestored, err
+		}
+
 		addRestoreAnnotation(obj)
 
 		if err := v.reconciler.Create(v.ctx, obj); err != nil {
@@ -2142,7 +2166,11 @@ func (v *VRGInstance) updateExistingPVForSync(pv *corev1.PersistentVolume) error
 	// failover/relocate process. Hence, the restore may not be
 	// required and the annotation for restore can be missing for
 	// the sync mode.
-	cleanupPVForRestore(pv)
+	err := v.cleanupPVForRestore(pv)
+	if err != nil {
+		return err
+	}
+
 	addRestoreAnnotation(pv)
 
 	if err := v.reconciler.Update(v.ctx, pv); err != nil {
@@ -2300,22 +2328,61 @@ func addRestoreAnnotation(obj client.Object) {
 	obj.GetAnnotations()[RestoreAnnotation] = RestoredByRamen
 }
 
+func SecretFromSC(params map[string]string,
+	secretName, secretNamespace string,
+) *corev1.SecretReference {
+	t := func(s string) string {
+		value, exists := params[s]
+		if exists {
+			return value
+		}
+
+		return ""
+	}
+
+	return &corev1.SecretReference{
+		Name:      t(secretName),
+		Namespace: t(secretNamespace),
+	}
+}
+
+func (v *VRGInstance) processPVSecrets(pv *corev1.PersistentVolume) error {
+	sc, err := v.getStorageClassFromSCName(&pv.Spec.StorageClassName)
+	if err != nil {
+		return err
+	}
+
+	pv.Spec.CSI.ControllerPublishSecretRef = SecretFromSC(sc.Parameters, controllerPublishSecretName,
+		controllerPublishSecretNamespace)
+	pv.Spec.CSI.NodeStageSecretRef = SecretFromSC(sc.Parameters, nodeStageSecretName, nodeStageSecretNamespace)
+	pv.Spec.CSI.NodePublishSecretRef = SecretFromSC(sc.Parameters, nodePublishSecretName, nodePublishSecretNamespace)
+	pv.Spec.CSI.ControllerExpandSecretRef = SecretFromSC(sc.Parameters, controllerExpandSecretName,
+		controllerExpandSecretNamespace)
+	pv.Spec.CSI.NodeExpandSecretRef = SecretFromSC(sc.Parameters, nodeExpandSecretName, nodeExpandSecretNamespace)
+
+	return nil
+}
+
 // cleanupForRestore cleans up required PV or PVC fields, to ensure restore succeeds
 // to a new cluster, and rebinding the PVC to an existing PV with the same claimRef
-func cleanupPVForRestore(pv *corev1.PersistentVolume) {
+func (v *VRGInstance) cleanupPVForRestore(pv *corev1.PersistentVolume) error {
 	pv.ResourceVersion = ""
 	if pv.Spec.ClaimRef != nil {
 		pv.Spec.ClaimRef.UID = ""
 		pv.Spec.ClaimRef.ResourceVersion = ""
 		pv.Spec.ClaimRef.APIVersion = ""
 	}
+
+	return v.processPVSecrets(pv)
 }
 
-func cleanupPVCForRestore(pvc *corev1.PersistentVolumeClaim) {
+func cleanupPVCForRestore(pvc *corev1.PersistentVolumeClaim) error {
 	pvc.ObjectMeta.Annotations = PruneAnnotations(pvc.GetAnnotations())
 	pvc.ObjectMeta.Finalizers = []string{}
 	pvc.ObjectMeta.ResourceVersion = ""
 	pvc.ObjectMeta.OwnerReferences = nil
+
+	return nil
 }
 
 // Follow this logic to update VRG (and also ProtectedPVC) conditions for VolRep