diff --git a/cmd/cephcsi.go b/cmd/cephcsi.go
index db5239fe9..71b3ce6ea 100644
--- a/cmd/cephcsi.go
+++ b/cmd/cephcsi.go
@@ -69,6 +69,7 @@ func init() {
 	flag.StringVar(&conf.PluginPath, "pluginpath", defaultPluginPath, "plugin path")
 	flag.StringVar(&conf.StagingPath, "stagingpath", defaultStagingPath, "staging path")
 	flag.StringVar(&conf.ClusterName, "clustername", "", "name of the cluster")
+	flag.BoolVar(&conf.SetMetadata, "setmetadata", false, "set metadata on the volume")
 	flag.StringVar(&conf.InstanceID, "instanceid", "", "Unique ID distinguishing this instance of Ceph CSI among other"+
 		" instances, when sharing Ceph clusters across CSI instances for provisioning")
 	flag.IntVar(&conf.PidLimit, "pidlimit", 0, "the PID limit to configure through cgroups")
@@ -251,6 +252,7 @@ func main() {
 			DriverName:  dname,
 			Namespace:   conf.DriverNamespace,
 			ClusterName: conf.ClusterName,
+			SetMetadata: conf.SetMetadata,
 		}
 		// initialize all controllers before starting.
 		initControllers()
diff --git a/internal/controller/controller.go b/internal/controller/controller.go
index bf608ca12..7bc6e157d 100644
--- a/internal/controller/controller.go
+++ b/internal/controller/controller.go
@@ -38,6 +38,7 @@ type Config struct {
 	DriverName  string
 	Namespace   string
 	ClusterName string
+	SetMetadata bool
 }
 
 // ControllerList holds the list of managers need to be started.
diff --git a/internal/controller/persistentvolume/persistentvolume.go b/internal/controller/persistentvolume/persistentvolume.go
index 5183fbb73..fe2ed5a60 100644
--- a/internal/controller/persistentvolume/persistentvolume.go
+++ b/internal/controller/persistentvolume/persistentvolume.go
@@ -185,6 +185,7 @@ func (r ReconcilePersistentVolume) reconcilePV(ctx context.Context, obj runtime.
 		requestName,
 		pvcNamespace,
 		r.config.ClusterName,
+		r.config.SetMetadata,
 		cr)
 	if err != nil {
 		log.ErrorLogMsg("failed to regenerate journal %s", err)
diff --git a/internal/rbd/controllerserver.go b/internal/rbd/controllerserver.go
index b67760ddb..b6a8f7e6a 100644
--- a/internal/rbd/controllerserver.go
+++ b/internal/rbd/controllerserver.go
@@ -55,6 +55,9 @@ type ControllerServer struct {
 
 	// Cluster name
 	ClusterName string
+
+	// Set metadata on volume
+	SetMetadata bool
 }
 
 func (cs *ControllerServer) validateVolumeReq(ctx context.Context, req *csi.CreateVolumeRequest) error {
@@ -173,7 +176,10 @@ func (cs *ControllerServer) parseVolCreateRequest(
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
 
+	// set cluster name on volume
 	rbdVol.ClusterName = cs.ClusterName
+	// set metadata on volume
+	rbdVol.EnableMetadata = cs.SetMetadata
 
 	// if the KMS is of type VaultToken, additional metadata is needed
 	// depending on the tenant, the KMS can be configured with other
@@ -1061,6 +1067,7 @@ func (cs *ControllerServer) CreateSnapshot(
 
 		return nil, err
 	}
+	rbdVol.EnableMetadata = cs.SetMetadata
 
 	// Check if source volume was created with required image features for snaps
 	if !rbdVol.hasSnapshotFeature() {
diff --git a/internal/rbd/driver/driver.go b/internal/rbd/driver/driver.go
index a74cd2552..5019d02cc 100644
--- a/internal/rbd/driver/driver.go
+++ b/internal/rbd/driver/driver.go
@@ -162,6 +162,7 @@ func (r *Driver) Run(conf *util.Config) {
 	if conf.IsControllerServer {
 		r.cs = NewControllerServer(r.cd)
 		r.cs.ClusterName = conf.ClusterName
+		r.cs.SetMetadata = conf.SetMetadata
 		r.rs = NewReplicationServer(r.cs)
 	}
 	if !conf.IsControllerServer && !conf.IsNodeServer {
diff --git a/internal/rbd/rbd_journal.go b/internal/rbd/rbd_journal.go
index 9da5b3ad8..eadfa99e7 100644
--- a/internal/rbd/rbd_journal.go
+++ b/internal/rbd/rbd_journal.go
@@ -543,6 +543,7 @@ func RegenerateJournal(
 	requestName,
 	owner,
 	clusterName string,
+	setMetadata bool,
 	cr *util.Credentials,
 ) (string, error) {
 	ctx := context.Background()
@@ -557,6 +558,7 @@ func RegenerateJournal(
 	rbdVol = &rbdVolume{}
 	rbdVol.VolID = volumeID
 	rbdVol.ClusterName = clusterName
+	rbdVol.EnableMetadata = setMetadata
 
 	err = vi.DecomposeCSIID(rbdVol.VolID)
 	if err != nil {
diff --git a/internal/rbd/rbd_util.go b/internal/rbd/rbd_util.go
index 5f26b2579..59c26f187 100644
--- a/internal/rbd/rbd_util.go
+++ b/internal/rbd/rbd_util.go
@@ -132,6 +132,8 @@ type rbdImage struct {
 
 	// Cluster name
 	ClusterName string
+	// Set metadata on volume
+	EnableMetadata bool
 
 	// encryption provides access to optional VolumeEncryption functions
 	encryption *util.VolumeEncryption
@@ -2061,6 +2063,10 @@ func genVolFromVolIDWithMigration(
 
 // setAllMetadata set all the metadata from arg parameters on RBD image.
 func (rv *rbdVolume) setAllMetadata(parameters map[string]string) error {
+	if !rv.EnableMetadata {
+		return nil
+	}
+
 	for k, v := range parameters {
 		err := rv.SetMetadata(k, v)
 		if err != nil {
@@ -2081,6 +2087,10 @@ func (rv *rbdVolume) setAllMetadata(parameters map[string]string) error {
 
 // unsetAllMetadata unset all the metadata from arg keys on RBD image.
 func (rv *rbdVolume) unsetAllMetadata(keys []string) error {
+	if !rv.EnableMetadata {
+		return nil
+	}
+
 	for _, key := range keys {
 		err := rv.RemoveMetadata(key)
 		// TODO: replace string comparison with errno.
diff --git a/internal/util/util.go b/internal/util/util.go
index f8d81a3d4..b9d7ea173 100644
--- a/internal/util/util.go
+++ b/internal/util/util.go
@@ -101,6 +101,8 @@ type Config struct {
 	// cephfs related flags
 	ForceKernelCephFS bool // force to use the ceph kernel client even if the kernel is < 4.17
 
+	SetMetadata bool // set metadata on the volume
+
 	// RbdHardMaxCloneDepth is the hard limit for maximum number of nested volume clones that are taken before a flatten
 	// occurs
 	RbdHardMaxCloneDepth uint