diff --git a/internal/rbd/rbd_util.go b/internal/rbd/rbd_util.go
index 4d4986ae2..c0cdfe796 100644
--- a/internal/rbd/rbd_util.go
+++ b/internal/rbd/rbd_util.go
@@ -28,6 +28,7 @@ import (
 	"time"
 
 	"github.com/ceph/ceph-csi/internal/util"
+	"github.com/ceph/ceph-csi/internal/util/k8s"
 	"github.com/ceph/ceph-csi/internal/util/log"
 
 	"github.com/ceph/go-ceph/rados"
@@ -1920,3 +1921,15 @@ func genVolFromVolIDWithMigration(
 
 	return rv, err
 }
+
+// setVolumeMetadata set PV/PVC/PVCNamespace metadata on RBD image.
+func (rv *rbdVolume) setVolumeMetadata(parameters map[string]string) error {
+	for k, v := range k8s.GetVolumeMetadata(parameters) {
+		err := rv.SetMetadata(k, v)
+		if err != nil {
+			return fmt.Errorf("failed to set metadata key %q, value %q on image: %w", k, v, err)
+		}
+	}
+
+	return nil
+}
diff --git a/internal/util/k8s/parameters.go b/internal/util/k8s/parameters.go
index 5d85182b5..e5f36cb6b 100644
--- a/internal/util/k8s/parameters.go
+++ b/internal/util/k8s/parameters.go
@@ -23,7 +23,12 @@ import (
 // to the driver on CreateVolumeRequest/CreateSnapshotRequest calls.
 const (
 	csiParameterPrefix = "csi.storage.k8s.io/"
-	pvcNamespaceKey    = "csi.storage.k8s.io/pvc/namespace"
+
+	// PV and PVC metadata keys used by external provisioner as part of
+	// create requests as parameters, when `extra-create-metadata` is true.
+	pvcNameKey      = "csi.storage.k8s.io/pvc/name"
+	pvcNamespaceKey = "csi.storage.k8s.io/pvc/namespace"
+	pvNameKey       = "csi.storage.k8s.io/pv/name"
 )
 
 // RemoveCSIPrefixedParameters removes parameters prefixed with csiParameterPrefix.
@@ -43,3 +48,18 @@ func RemoveCSIPrefixedParameters(param map[string]string) map[string]string {
 func GetOwner(param map[string]string) string {
 	return param[pvcNamespaceKey]
 }
+
+// GetVolumeMetadata filter parameters, only return PV/PVC/PVCNamespace metadata.
+func GetVolumeMetadata(parameters map[string]string) map[string]string {
+	keys := []string{pvcNameKey, pvcNamespaceKey, pvNameKey}
+	newParam := map[string]string{}
+	for k, v := range parameters {
+		for _, key := range keys {
+			if strings.Contains(k, key) {
+				newParam[k] = v
+			}
+		}
+	}
+
+	return newParam
+}