diff --git a/charts/weka-operator/templates/manager.yaml b/charts/weka-operator/templates/manager.yaml
index 1209a35a8..c179e00d0 100644
--- a/charts/weka-operator/templates/manager.yaml
+++ b/charts/weka-operator/templates/manager.yaml
@@ -272,6 +272,8 @@ spec:
               value: "{{ .Values.csi.preventNewWorkloadOnClientContainerNotRunning }}"
             - name: CSI_LOG_LEVEL
               value: "{{ .Values.csi.logLevel }}"
+            - name: CSI_SET_OWNERSHIP_ON_DYNAMIC_FILESYSTEMS
+              value: "{{ .Values.csi.setOwnershipOnDynamicFilesystems }}"
             - name: CSI_CONTROLLER_WEKAFS_LIMITS_CPU
               value: "{{ .Values.csi.controller.resources.wekafs.limits.cpu }}"
             - name: CSI_CONTROLLER_WEKAFS_LIMITS_MEMORY
diff --git a/charts/weka-operator/values.yaml b/charts/weka-operator/values.yaml
index 54959bdbf..b4a9d0aed 100644
--- a/charts/weka-operator/values.yaml
+++ b/charts/weka-operator/values.yaml
@@ -270,6 +270,7 @@ csi:
   registrarImage: "registry.k8s.io/sig-storage/csi-node-driver-registrar:v2.14.0"
   preventNewWorkloadOnClientContainerNotRunning: true
   logLevel: 5
+  setOwnershipOnDynamicFilesystems: false
   controller:
     resources:
       wekafs:
diff --git a/internal/config/env.go b/internal/config/env.go
index 8a92fb264..b45a21a76 100644
--- a/internal/config/env.go
+++ b/internal/config/env.go
@@ -128,6 +128,7 @@ type EmbeddedCsiSettings struct {
 	RegistrarImage                                string
 	PreventNewWorkloadOnClientContainerNotRunning bool
 	LogLevel                                      int
+	SetOwnershipOnDynamicFilesystems              bool
 	ControllerResources                           CsiControllerResources
 	NodeResources                                 CsiNodeResources
 }
@@ -389,6 +390,7 @@ func ConfigureEnv(ctx context.Context) {
 	Config.Csi.RegistrarImage = env.GetString("CSI_REGISTRAR_IMAGE", "")
 	Config.Csi.PreventNewWorkloadOnClientContainerNotRunning = getBoolEnvOrDefault("CSI_PREVENT_NEW_WORKLOAD_ON_CLIENT_CONTAINER_NOT_RUNNING", true)
 	Config.Csi.LogLevel = getIntEnvOrDefault("CSI_LOG_LEVEL", 5)
+	Config.Csi.SetOwnershipOnDynamicFilesystems = getBoolEnvOrDefault("CSI_SET_OWNERSHIP_ON_DYNAMIC_FILESYSTEMS", false)
 	Config.Csi.ControllerResources = parseCsiControllerResources()
 	Config.Csi.NodeResources = parseCsiNodeResources()
 	Config.SyslogPackage = getEnvOrDefault("SYSLOG_PACKAGE", "auto")
diff --git a/internal/controllers/operations/csi/controller.go b/internal/controllers/operations/csi/controller.go
index ca73b1bee..655ce7845 100644
--- a/internal/controllers/operations/csi/controller.go
+++ b/internal/controllers/operations/csi/controller.go
@@ -23,21 +23,22 @@ import (
 // CsiControllerHashableSpec represents the fields from CSI Controller Deployment
 // that are relevant for determining if an update is needed
 type CsiControllerHashableSpec struct {
-	CsiDriverName         string
-	CsiImage              string
-	CsiAttacherImage      string
-	CsiProvisionerImage   string
-	CsiResizerImage       string
-	CsiSnapshotterImage   string
-	CsiLivenessProbeImage string
-	Labels                *util2.HashableMap
-	Tolerations           []corev1.Toleration
-	NodeSelector          *util2.HashableMap
-	EnforceTrustedHttps   bool
-	SkipGarbageCollection bool
-	LogLevel              int
-	PriorityClassName     string
-	WekaContainerName     string
+	CsiDriverName                       string
+	CsiImage                            string
+	CsiAttacherImage                    string
+	CsiProvisionerImage                 string
+	CsiResizerImage                     string
+	CsiSnapshotterImage                 string
+	CsiLivenessProbeImage               string
+	Labels                              *util2.HashableMap
+	Tolerations                         []corev1.Toleration
+	NodeSelector                        *util2.HashableMap
+	EnforceTrustedHttps                 bool
+	SkipGarbageCollection               bool
+	SetOwnershipOnDynamicFilesystems    bool
+	LogLevel                            int
+	PriorityClassName                   string
+	WekaContainerName                   string
 }
 
 // GetCsiControllerDeploymentHash generates a hash for the CSI Controller Deployment
@@ -68,21 +69,22 @@ func GetCsiControllerDeploymentHash(csiGroupName string, wekaClient *weka.WekaCl
 	}
 
 	spec := CsiControllerHashableSpec{
-		CsiDriverName:         csiDriverName,
-		CsiImage:              config.Config.Csi.WekafsImage,
-		CsiAttacherImage:      config.Config.Csi.AttacherImage,
-		CsiProvisionerImage:   config.Config.Csi.ProvisionerImage,
-		CsiResizerImage:       config.Config.Csi.ResizerImage,
-		CsiSnapshotterImage:   config.Config.Csi.SnapshotterImage,
-		CsiLivenessProbeImage: config.Config.Csi.LivenessProbeImage,
-		Labels:                labelsHashable,
-		Tolerations:           tolerations,
-		NodeSelector:          nodeSelectorHashable,
-		EnforceTrustedHttps:   enforceTrustedHttps,
-		SkipGarbageCollection: skipGarbageCollection,
-		LogLevel:              config.Config.Csi.LogLevel,
-		PriorityClassName:     config.Config.PriorityClasses.Targeted,
-		WekaContainerName:     resources.GetWekaClientContainerName(wekaClient),
+		CsiDriverName:                    csiDriverName,
+		CsiImage:                         config.Config.Csi.WekafsImage,
+		CsiAttacherImage:                 config.Config.Csi.AttacherImage,
+		CsiProvisionerImage:              config.Config.Csi.ProvisionerImage,
+		CsiResizerImage:                  config.Config.Csi.ResizerImage,
+		CsiSnapshotterImage:              config.Config.Csi.SnapshotterImage,
+		CsiLivenessProbeImage:            config.Config.Csi.LivenessProbeImage,
+		Labels:                           labelsHashable,
+		Tolerations:                      tolerations,
+		NodeSelector:                     nodeSelectorHashable,
+		EnforceTrustedHttps:              enforceTrustedHttps,
+		SkipGarbageCollection:            skipGarbageCollection,
+		SetOwnershipOnDynamicFilesystems: config.Config.Csi.SetOwnershipOnDynamicFilesystems,
+		LogLevel:                         config.Config.Csi.LogLevel,
+		PriorityClassName:                config.Config.PriorityClasses.Targeted,
+		WekaContainerName:                resources.GetWekaClientContainerName(wekaClient),
 	}
 
 	return util2.HashStruct(spec)
@@ -163,6 +165,9 @@ func NewCsiControllerDeployment(ctx context.Context, csiGroupName string, wekaCl
 	if skipGarbageCollection {
 		args = append(args, "--skipgarbagecollection")
 	}
+	if config.Config.Csi.SetOwnershipOnDynamicFilesystems {
+		args = append(args, "--setownershipondynamicfilesystems")
+	}
 
 	tracingFlag := GetTracingFlag()
 	if tracingFlag != "" {