diff --git a/Documentation/Upgrade/ceph-upgrade.md b/Documentation/Upgrade/ceph-upgrade.md
index 1efa5535f..1cca31cad 100644
--- a/Documentation/Upgrade/ceph-upgrade.md
+++ b/Documentation/Upgrade/ceph-upgrade.md
@@ -28,6 +28,12 @@ The following Ceph versions are supported in this release of Rook:
 
 * Ceph Squid v19.2.0 or newer
 * Ceph Tentacle v20.2.0 or newer
+    * IMPORTANT: **There is a known data corruption issue in v20.2.0**, if the "read affinity" feature is enabled.
+    Read affinity is disabled by default, and is enabled by the CephCluster setting `csi.readAffinity.enabled: true`.
+    If you have enabled read affinity, we recommend waiting for v20.2.1 before upgrading to Ceph Tentacle v20.
+    If read affinity is enabled with v20.2.0, Rook automatically disables read affinity internally to help avoid this corruption.
+    After the upgrade, existing applications must be restarted to fully disable the read affinity in existing volumes.
+    See [this issue](https://github.com/rook/rook/issues/16839)) for more details.
 
 !!! important
     When an update is requested, the operator will check Ceph's status,
diff --git a/pkg/operator/ceph/csi/ceph_connection.go b/pkg/operator/ceph/csi/ceph_connection.go
index 30de5cb86..be7c068cd 100644
--- a/pkg/operator/ceph/csi/ceph_connection.go
+++ b/pkg/operator/ceph/csi/ceph_connection.go
@@ -25,6 +25,7 @@ import (
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd/topology"
+	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 
 	csiopv1 "github.com/ceph/ceph-csi-operator/api/v1"
@@ -73,7 +74,7 @@ func CreateUpdateCephConnection(c client.Client, clusterInfo *cephclient.Cluster
 func generateCephConnSpec(c client.Client, clusterInfo *cephclient.ClusterInfo, clusterSpec cephv1.ClusterSpec) (csiopv1.CephConnectionSpec, error) {
 	csiClusterConnSpec := csiopv1.CephConnectionSpec{}
 
-	if clusterSpec.CSI.ReadAffinity.Enabled {
+	if ReadAffinityEnabled(clusterSpec.CSI.ReadAffinity.Enabled, clusterInfo.CephVersion) {
 		crushLabels := clusterSpec.CSI.ReadAffinity.CrushLocationLabels
 		if len(crushLabels) == 0 {
 			logger.Debug("using default crush topology labels")
@@ -100,3 +101,16 @@ func generateCephConnSpec(c client.Client, clusterInfo *cephclient.ClusterInfo,
 	csiClusterConnSpec.Monitors = MonEndpoints(clusterInfo.AllMonitors(), clusterSpec.RequireMsgr2())
 	return csiClusterConnSpec, nil
 }
+
+func ReadAffinityEnabled(desiredEnabled bool, version cephver.CephVersion) bool {
+	if !desiredEnabled {
+		return false
+	}
+	logger.Debugf("checking if read affinity is supported for Ceph version %q", version.String())
+	// Read affinity causes corruption on Ceph v20.2.0, so always disable it for that version
+	if version.Major == 20 && version.Minor == 2 && version.Extra == 0 {
+		logger.Warningf("read affinity disabled for Ceph version v20.2.0")
+		return false
+	}
+	return true
+}
diff --git a/pkg/operator/ceph/csi/ceph_connection_test.go b/pkg/operator/ceph/csi/ceph_connection_test.go
index 2255be4e4..c4ad1a2c8 100644
--- a/pkg/operator/ceph/csi/ceph_connection_test.go
+++ b/pkg/operator/ceph/csi/ceph_connection_test.go
@@ -26,6 +26,7 @@ import (
 	"github.com/rook/rook/pkg/client/clientset/versioned/scheme"
 	clienttest "github.com/rook/rook/pkg/daemon/ceph/client/test"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd/topology"
+	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	"github.com/stretchr/testify/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -149,3 +150,50 @@ func TestCephConnectionDefaultTopology(t *testing.T) {
 		assert.True(t, labels[defaultLabel], "expected label %q not found", defaultLabel)
 	}
 }
+
+func TestReadAffinityEnabled(t *testing.T) {
+	tests := []struct {
+		name         string
+		enabledField bool
+		cephVersion  cephver.CephVersion
+		expected     bool
+	}{
+		{
+			name:         "enabled field false, ceph version < 20.2.0",
+			enabledField: false,
+			cephVersion:  cephver.CephVersion{Major: 19, Minor: 2, Extra: 0},
+			expected:     false,
+		},
+		{
+			name:         "enabled field true, ceph version < 20.2.0",
+			enabledField: true,
+			cephVersion:  cephver.CephVersion{Major: 19, Minor: 2, Extra: 0},
+			expected:     true,
+		},
+		{
+			name:         "enabled field true, ceph version == 20.2.0",
+			enabledField: true,
+			cephVersion:  cephver.CephVersion{Major: 20, Minor: 2, Extra: 0},
+			expected:     false,
+		},
+		{
+			name:         "enabled field true, ceph version > 20.2.0",
+			enabledField: true,
+			cephVersion:  cephver.CephVersion{Major: 20, Minor: 2, Extra: 1},
+			expected:     true,
+		},
+		{
+			name:         "enabled field false, ceph version > 20.2.0",
+			enabledField: false,
+			cephVersion:  cephver.CephVersion{Major: 20, Minor: 2, Extra: 1},
+			expected:     false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := ReadAffinityEnabled(tt.enabledField, tt.cephVersion)
+			assert.Equal(t, tt.expected, result)
+		})
+	}
+}
diff --git a/pkg/operator/ceph/csi/cluster_config.go b/pkg/operator/ceph/csi/cluster_config.go
index b4531ff6f..855a77d05 100644
--- a/pkg/operator/ceph/csi/cluster_config.go
+++ b/pkg/operator/ceph/csi/cluster_config.go
@@ -148,7 +148,7 @@ func updateNetNamespaceFilePath(clusterNamespace string, cc csiClusterConfig) {
 
 // updateCsiClusterConfig returns a json-formatted string containing
 // the cluster-to-mon mapping required to configure ceph csi.
-func updateCsiClusterConfig(curr, clusterID, clusterNamespace string, newCsiClusterConfigEntry *CSIClusterConfigEntry) (string, error) {
+func updateCsiClusterConfig(curr, clusterID string, clusterInfo *cephclient.ClusterInfo, newCsiClusterConfigEntry *CSIClusterConfigEntry) (string, error) {
 	var (
 		cc     csiClusterConfig
 		centry CSIClusterConfigEntry
@@ -165,12 +165,17 @@ func updateCsiClusterConfig(curr, clusterID, clusterNamespace string, newCsiClus
 	// wait for the other update to complete. Monitors and Subvolumegroup will be updated
 	// independently and won't collide.
 	if newCsiClusterConfigEntry != nil {
+		// Disable read affinity if needed for the ceph version
+		if !ReadAffinityEnabled(newCsiClusterConfigEntry.ReadAffinity.Enabled, clusterInfo.CephVersion) {
+			newCsiClusterConfigEntry.ReadAffinity = cephcsi.ReadAffinity{Enabled: false}
+		}
+
 		for i, centry := range cc {
 			// there is a bug with an unknown source where the csi config can have an empty
 			// namespace entry. detect and fix this scenario if it occurs
 			if centry.Namespace == "" && centry.ClusterID == clusterID {
-				logger.Infof("correcting CSI config map entry for cluster ID %q; empty namespace will be set to %q", clusterID, clusterNamespace)
-				centry.Namespace = clusterNamespace
+				logger.Infof("correcting CSI config map entry for cluster ID %q; empty namespace will be set to %q", clusterID, clusterInfo.Namespace)
+				centry.Namespace = clusterInfo.Namespace
 				cc[i] = centry
 			}
 
@@ -223,7 +228,7 @@ func updateCsiClusterConfig(curr, clusterID, clusterNamespace string, newCsiClus
 		// will fail with a dangling pointer
 		if newCsiClusterConfigEntry != nil {
 			centry.ClusterID = clusterID
-			centry.Namespace = clusterNamespace
+			centry.Namespace = clusterInfo.Namespace
 			centry.Monitors = newCsiClusterConfigEntry.Monitors
 			centry.RBD = newCsiClusterConfigEntry.RBD
 			centry.CephFS = newCsiClusterConfigEntry.CephFS
@@ -328,7 +333,7 @@ func SaveClusterConfig(clientset kubernetes.Interface, clusterID, clusterNamespa
 
 	if newCsiClusterConfigEntry != nil {
 		// set CSIDriverOptions
-		newCsiClusterConfigEntry.ReadAffinity.Enabled = clusterInfo.CSIDriverSpec.ReadAffinity.Enabled
+		newCsiClusterConfigEntry.ReadAffinity.Enabled = ReadAffinityEnabled(clusterInfo.CSIDriverSpec.ReadAffinity.Enabled, clusterInfo.CephVersion)
 		newCsiClusterConfigEntry.ReadAffinity.CrushLocationLabels = clusterInfo.CSIDriverSpec.ReadAffinity.CrushLocationLabels
 
 		newCsiClusterConfigEntry.CephFS.KernelMountOptions = clusterInfo.CSIDriverSpec.CephFS.KernelMountOptions
@@ -353,7 +358,7 @@ func SaveClusterConfig(clientset kubernetes.Interface, clusterID, clusterNamespa
 		currData = "[]"
 	}
 
-	newData, err := updateCsiClusterConfig(currData, clusterID, clusterNamespace, newCsiClusterConfigEntry)
+	newData, err := updateCsiClusterConfig(currData, clusterID, clusterInfo, newCsiClusterConfigEntry)
 	if err != nil {
 		return errors.Wrap(err, "failed to update csi config map data")
 	}
@@ -369,7 +374,7 @@ func SaveClusterConfig(clientset kubernetes.Interface, clusterID, clusterNamespa
 
 // updateCSIDriverOptions updates the CSI driver options, including read affinity, kernel mount options
 // and fuse mount options, for all entries belonging to the same cluster.
-func updateCSIDriverOptions(curr, clusterKey string,
+func updateCSIDriverOptions(curr string, clusterInfo *cephclient.ClusterInfo,
 	csiDriverOptions *cephv1.CSIDriverSpec,
 ) (string, error) {
 	cc, err := parseCsiClusterConfig(curr)
@@ -379,8 +384,8 @@ func updateCSIDriverOptions(curr, clusterKey string,
 
 	for i := range cc {
 		// If the clusterID belongs to the same cluster, update the entry.
-		if clusterKey == cc[i].Namespace {
-			cc[i].ReadAffinity.Enabled = csiDriverOptions.ReadAffinity.Enabled
+		if clusterInfo.Namespace == cc[i].Namespace {
+			cc[i].ReadAffinity.Enabled = ReadAffinityEnabled(csiDriverOptions.ReadAffinity.Enabled, clusterInfo.CephVersion)
 			cc[i].ReadAffinity.CrushLocationLabels = csiDriverOptions.ReadAffinity.CrushLocationLabels
 
 			cc[i].CephFS.KernelMountOptions = csiDriverOptions.CephFS.KernelMountOptions
@@ -388,7 +393,7 @@ func updateCSIDriverOptions(curr, clusterKey string,
 		}
 	}
 
-	updateNetNamespaceFilePath(clusterKey, cc)
+	updateNetNamespaceFilePath(clusterInfo.Namespace, cc)
 	return formatCsiClusterConfig(cc)
 }
 
@@ -423,7 +428,7 @@ func SaveCSIDriverOptions(clientset kubernetes.Interface, clusterNamespace strin
 		currData = "[]"
 	}
 
-	newData, err := updateCSIDriverOptions(currData, clusterNamespace, &clusterInfo.CSIDriverSpec)
+	newData, err := updateCSIDriverOptions(currData, clusterInfo, &clusterInfo.CSIDriverSpec)
 	if err != nil {
 		return errors.Wrap(err, "failed to update csi config map data")
 	}
diff --git a/pkg/operator/ceph/csi/cluster_config_test.go b/pkg/operator/ceph/csi/cluster_config_test.go
index 23c78d85a..0954a970b 100644
--- a/pkg/operator/ceph/csi/cluster_config_test.go
+++ b/pkg/operator/ceph/csi/cluster_config_test.go
@@ -118,9 +118,10 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 
 	var want, s string
 	var err error
+	clusterInfo := &cephclient.ClusterInfo{Namespace: "rook-ceph-1"}
 
 	t.Run("add a simple mons list", func(t *testing.T) {
-		s, err = updateCsiClusterConfig("[]", "rook-ceph-1", "rook-ceph-1", &csiClusterConfigEntry)
+		s, err = updateCsiClusterConfig("[]", "rook-ceph-1", clusterInfo, &csiClusterConfigEntry)
 		assert.NoError(t, err)
 		want = `[{"clusterID":"rook-ceph-1","monitors":["1.2.3.4:5000"],"namespace":"rook-ceph-1"}]`
 		compareJSON(t, want, s)
@@ -128,7 +129,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 
 	t.Run("add a 2nd mon to the current cluster", func(t *testing.T) {
 		csiClusterConfigEntry.Monitors = append(csiClusterConfigEntry.Monitors, "10.11.12.13:5000")
-		s, err = updateCsiClusterConfig(s, "rook-ceph-1", "rook-ceph-1", &csiClusterConfigEntry)
+		s, err = updateCsiClusterConfig(s, "rook-ceph-1", clusterInfo, &csiClusterConfigEntry)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -140,7 +141,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	})
 
 	t.Run("add mount options to the current cluster", func(t *testing.T) {
-		configWithMountOptions, err := updateCsiClusterConfig(s, "rook-ceph-1", "rook-ceph-1", &csiClusterConfigEntryMountOptions)
+		configWithMountOptions, err := updateCsiClusterConfig(s, "rook-ceph-1", clusterInfo, &csiClusterConfigEntryMountOptions)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(configWithMountOptions)
 		assert.NoError(t, err)
@@ -151,7 +152,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	})
 
 	t.Run("add a 2nd cluster with 3 mons", func(t *testing.T) {
-		s, err = updateCsiClusterConfig(s, "beta", "beta", &csiClusterConfigEntry2)
+		clusterInfo.Namespace = "beta"
+		s, err = updateCsiClusterConfig(s, "beta", clusterInfo, &csiClusterConfigEntry2)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -174,8 +176,9 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	t.Run("remove a mon from the 2nd cluster", func(t *testing.T) {
 		i := 2
 		// Remove last element of the slice
+		clusterInfo.Namespace = "beta"
 		csiClusterConfigEntry2.Monitors = append(csiClusterConfigEntry2.Monitors[:i], csiClusterConfigEntry2.Monitors[i+1:]...)
-		s, err = updateCsiClusterConfig(s, "beta", "beta", &csiClusterConfigEntry2)
+		s, err = updateCsiClusterConfig(s, "beta", clusterInfo, &csiClusterConfigEntry2)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -191,7 +194,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	})
 
 	t.Run("add a 3rd cluster with subvolumegroup", func(t *testing.T) {
-		s, err = updateCsiClusterConfig(s, "baba", "baba", &csiClusterConfigEntry3)
+		clusterInfo.Namespace = "baba"
+		s, err = updateCsiClusterConfig(s, "baba", clusterInfo, &csiClusterConfigEntry3)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -206,7 +210,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	})
 
 	t.Run("update mount options in presence of subvolumegroup", func(t *testing.T) {
-		sMntOptionUpdate, err := updateCsiClusterConfig(s, "baba", "baba", &csiClusterConfigEntryMountOptions)
+		clusterInfo.Namespace = "baba"
+		sMntOptionUpdate, err := updateCsiClusterConfig(s, "baba", clusterInfo, &csiClusterConfigEntryMountOptions)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(sMntOptionUpdate)
 		assert.NoError(t, err)
@@ -218,8 +223,9 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	})
 
 	t.Run("add a 4th mon to the 3rd cluster and subvolumegroup is preserved", func(t *testing.T) {
+		clusterInfo.Namespace = "baba"
 		csiClusterConfigEntry3.Monitors = append(csiClusterConfigEntry3.Monitors, "10.11.12.13:5000")
-		s, err = updateCsiClusterConfig(s, "baba", "baba", &csiClusterConfigEntry3)
+		s, err = updateCsiClusterConfig(s, "baba", clusterInfo, &csiClusterConfigEntry3)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -230,7 +236,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 
 	t.Run("remove subvolumegroup", func(t *testing.T) {
 		csiClusterConfigEntry3.CephFS.SubvolumeGroup = ""
-		s, err = updateCsiClusterConfig(s, "baba", "baba", nil)
+		clusterInfo.Namespace = "baba"
+		s, err = updateCsiClusterConfig(s, "baba", clusterInfo, nil)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -245,7 +252,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 				},
 			},
 		}
-		s, err = updateCsiClusterConfig(s, "quatre", "quatre", &csiClusterConfigEntry4)
+		clusterInfo.Namespace = "quatre"
+		s, err = updateCsiClusterConfig(s, "quatre", clusterInfo, &csiClusterConfigEntry4)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -254,7 +262,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 		assert.Equal(t, "my-group2", cc[2].CephFS.SubvolumeGroup, cc)
 
 		csiClusterConfigEntry4.Monitors = []string{"10.1.1.1:5000", "10.1.1.2:5000", "10.1.1.3:5000"}
-		s, err = updateCsiClusterConfig(s, "quatre", "quatre", &csiClusterConfigEntry4)
+		s, err = updateCsiClusterConfig(s, "quatre", clusterInfo, &csiClusterConfigEntry4)
 		assert.NoError(t, err)
 		cc, err = parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -263,7 +271,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	})
 
 	t.Run("does it return error on garbage input?", func(t *testing.T) {
-		_, err = updateCsiClusterConfig("qqq", "beta", "beta", &csiClusterConfigEntry2)
+		clusterInfo.Namespace = "beta"
+		_, err = updateCsiClusterConfig("qqq", "beta", clusterInfo, &csiClusterConfigEntry2)
 		assert.Error(t, err)
 	})
 
@@ -278,7 +287,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 				Monitors: []string{"1.2.3.4:5000"},
 			},
 		}
-		s, err := updateCsiClusterConfig("[]", clusterIDofCluster1, clusterIDofCluster1, &csiCluster1ConfigEntry)
+		clusterInfo.Namespace = clusterIDofCluster1
+		s, err := updateCsiClusterConfig("[]", clusterIDofCluster1, clusterInfo, &csiCluster1ConfigEntry)
 		assert.NoError(t, err)
 		want = `[{"clusterID":"rook-ceph","monitors":["1.2.3.4:5000"],"namespace":"rook-ceph"}]`
 		compareJSON(t, want, s)
@@ -292,7 +302,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 				},
 			},
 		}
-		s, err = updateCsiClusterConfig(s, subvolGrpNameofCluster1, clusterIDofCluster1, &subVolCsiCluster1Config)
+		s, err = updateCsiClusterConfig(s, subvolGrpNameofCluster1, clusterInfo, &subVolCsiCluster1Config)
 		assert.NoError(t, err)
 		cc, err := parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -310,7 +320,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 				},
 			},
 		}
-		s, err = updateCsiClusterConfig(s, radosNSofCluster1, clusterIDofCluster1, &radosNsCsiCluster1Config)
+		s, err = updateCsiClusterConfig(s, radosNSofCluster1, clusterInfo, &radosNsCsiCluster1Config)
 		assert.NoError(t, err)
 		cc, err = parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -320,7 +330,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 
 		// update mon IP's and check is it updating for all clusterID's
 		csiCluster1ConfigEntry.Monitors = append(csiCluster1ConfigEntry.Monitors, "1.2.3.10:5000")
-		s, err = updateCsiClusterConfig(s, clusterIDofCluster1, clusterIDofCluster1, &csiCluster1ConfigEntry)
+		s, err = updateCsiClusterConfig(s, clusterIDofCluster1, clusterInfo, &csiCluster1ConfigEntry)
 		assert.NoError(t, err)
 		cc, err = parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -359,11 +369,12 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 				},
 			},
 		}
-		s, err = updateCsiClusterConfig(s, clusterIDofCluster2, clusterIDofCluster2, &csiCluster2ConfigEntry)
+		clusterInfo.Namespace = clusterIDofCluster2
+		s, err = updateCsiClusterConfig(s, clusterIDofCluster2, clusterInfo, &csiCluster2ConfigEntry)
 		assert.NoError(t, err)
-		s, err = updateCsiClusterConfig(s, subvolGrpNameofCluster2, clusterIDofCluster2, &subVolCsiCluster2Config)
+		s, err = updateCsiClusterConfig(s, subvolGrpNameofCluster2, clusterInfo, &subVolCsiCluster2Config)
 		assert.NoError(t, err)
-		s, err = updateCsiClusterConfig(s, radosNSofCluster2, clusterIDofCluster2, &radosNsCsiCluster2Config)
+		s, err = updateCsiClusterConfig(s, radosNSofCluster2, clusterInfo, &radosNsCsiCluster2Config)
 		assert.NoError(t, err)
 		cc, err = parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -382,7 +393,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 		// update mon on 2nd cluster and check is it updating for all clusterID's
 		// of 2nd cluster
 		csiCluster2ConfigEntry.Monitors = append(csiCluster2ConfigEntry.Monitors, "192.168.0.3:5000")
-		s, err = updateCsiClusterConfig(s, clusterIDofCluster2, clusterIDofCluster2, &csiCluster2ConfigEntry)
+		s, err = updateCsiClusterConfig(s, clusterIDofCluster2, clusterInfo, &csiCluster2ConfigEntry)
 		assert.NoError(t, err)
 		cc, err = parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -404,7 +415,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 		// clusterID's of 2nd cluster
 		i := 1
 		csiCluster2ConfigEntry.Monitors = append(csiCluster2ConfigEntry.Monitors[:i], csiCluster2ConfigEntry.Monitors[i+1:]...)
-		s, err = updateCsiClusterConfig(s, clusterIDofCluster2, clusterIDofCluster2, &csiCluster2ConfigEntry)
+		s, err = updateCsiClusterConfig(s, clusterIDofCluster2, clusterInfo, &csiCluster2ConfigEntry)
 		assert.NoError(t, err)
 		cc, err = parseCsiClusterConfig(s)
 		assert.NoError(t, err)
@@ -423,13 +434,15 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	})
 
 	t.Run("test multus cluster", func(t *testing.T) {
-		s, err = updateCsiClusterConfig("[]", "rook-ceph-1", "rook-ceph-1", &csiClusterConfigEntryMultus)
+		clusterInfo.Namespace = "rook-ceph-1"
+		s, err = updateCsiClusterConfig("[]", "rook-ceph-1", clusterInfo, &csiClusterConfigEntryMultus)
 		assert.NoError(t, err)
 		compareJSON(t, `[{"clusterID":"rook-ceph-1","monitors":["1.2.3.4:5000"],"rbd":{"netNamespaceFilePath":"/var/run/netns/rook-ceph-1","radosNamespace":"rook-ceph-1"},"namespace":"rook-ceph-1"}]`, s)
 	})
 
 	t.Run("test crush location labels are set", func(t *testing.T) {
-		s, err = updateCsiClusterConfig("[]", "rook-ceph-4", "rook-ceph-4", &csiClusterConfigEntry4)
+		clusterInfo.Namespace = "rook-ceph-4"
+		s, err = updateCsiClusterConfig("[]", "rook-ceph-4", clusterInfo, &csiClusterConfigEntry4)
 		assert.NoError(t, err)
 		compareJSON(t, `[{"clusterID":"rook-ceph-4","monitors":["10.1.1.1:5000"],"readAffinity": {"enabled": true, "crushLocationLabels":["kubernetes.io/hostname",
 		"topology.kubernetes.io/region","topology.kubernetes.io/zone","topology.rook.io/chassis","topology.rook.io/rack","topology.rook.io/row","topology.rook.io/pdu",
@@ -450,7 +463,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 			},
 		}
 		currentConfigWithoutNamespace := fmt.Sprintf(currentConfigFormatString, "")
-		out, err := updateCsiClusterConfig(currentConfigWithoutNamespace, idAndNs, idAndNs, csiConfigEntry)
+		clusterInfo.Namespace = idAndNs
+		out, err := updateCsiClusterConfig(currentConfigWithoutNamespace, idAndNs, clusterInfo, csiConfigEntry)
 		assert.NoError(t, err)
 		expectedOutput := fmt.Sprintf(currentConfigFormatString, idAndNs)
 		assert.Equal(t, expectedOutput, out)
@@ -473,7 +487,8 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 			},
 		}
 		currentConfigWithoutNamespace := fmt.Sprintf(currentConfigFormatString, cephFsNetNsFilePath, rbdNetNsFilePath, "")
-		out, err := updateCsiClusterConfig(currentConfigWithoutNamespace, idAndNs, idAndNs, csiConfigEntry)
+		clusterInfo.Namespace = idAndNs
+		out, err := updateCsiClusterConfig(currentConfigWithoutNamespace, idAndNs, clusterInfo, csiConfigEntry)
 		assert.NoError(t, err)
 		expectedOutput := fmt.Sprintf(currentConfigFormatString, "", "", idAndNs)
 		assert.Equal(t, expectedOutput, out)
@@ -700,7 +715,7 @@ func TestUpdateCSIDriverOptions(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			dataString, err := formatCsiClusterConfig(tt.args.clusterConfig)
 			assert.NoError(t, err)
-			got, err := updateCSIDriverOptions(dataString, tt.args.clusterKey, tt.args.csiDriverOptions)
+			got, err := updateCSIDriverOptions(dataString, &cephclient.ClusterInfo{Namespace: tt.args.clusterKey}, tt.args.csiDriverOptions)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("updateCSIDriverOptions() error = %v, wantErr %v", err, tt.wantErr)
 				return
diff --git a/pkg/operator/ceph/file/subvolumegroup/controller.go b/pkg/operator/ceph/file/subvolumegroup/controller.go
index 0731fc9d8..4f7b00f49 100644
--- a/pkg/operator/ceph/file/subvolumegroup/controller.go
+++ b/pkg/operator/ceph/file/subvolumegroup/controller.go
@@ -40,6 +40,7 @@ import (
 	"github.com/coreos/pkg/capnslog"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
+	"github.com/rook/rook/pkg/operator/ceph/config"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/csi"
 	"github.com/rook/rook/pkg/operator/ceph/file"
@@ -261,6 +262,13 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) reconcile(request reconcile.Requ
 		return reconcile.Result{}, nil
 	}
 
+	// Detect running Ceph version
+	runningCephVersion, err := cephclient.LeastUptodateDaemonVersion(r.context, r.clusterInfo, config.OsdType)
+	if err != nil {
+		return reconcile.Result{}, errors.Wrapf(err, "failed to retrieve current ceph %q version", config.OsdType)
+	}
+	r.clusterInfo.CephVersion = runningCephVersion
+
 	cephFilesystemSubVolumeGroupName := cephFilesystemSubVolumeGroup.Name
 	if cephFilesystemSubVolumeGroup.Spec.Name != "" {
 		cephFilesystemSubVolumeGroupName = cephFilesystemSubVolumeGroup.Spec.Name
@@ -358,7 +366,7 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) updateClusterConfig(cephFilesyst
 				FuseMountOptions:   r.clusterInfo.CSIDriverSpec.CephFS.FuseMountOptions,
 			},
 			ReadAffinity: cephcsi.ReadAffinity{
-				Enabled:             r.clusterInfo.CSIDriverSpec.ReadAffinity.Enabled,
+				Enabled:             csi.ReadAffinityEnabled(r.clusterInfo.CSIDriverSpec.ReadAffinity.Enabled, r.clusterInfo.CephVersion),
 				CrushLocationLabels: r.clusterInfo.CSIDriverSpec.ReadAffinity.CrushLocationLabels,
 			},
 		},
diff --git a/pkg/operator/ceph/file/subvolumegroup/controller_test.go b/pkg/operator/ceph/file/subvolumegroup/controller_test.go
index 674ffe0a7..78524c48f 100644
--- a/pkg/operator/ceph/file/subvolumegroup/controller_test.go
+++ b/pkg/operator/ceph/file/subvolumegroup/controller_test.go
@@ -77,11 +77,22 @@ func TestFilesystemSubvolumeGroupController(t *testing.T) {
 		cephFilesystemSubVolumeGroup,
 	}
 
+	cephDaemonVersions := `{
+	"mon": {
+		"ceph version 20.2.1 (0000000000000000) tentacle (stable)": 3
+	},
+	"osd": {
+		"ceph version 20.2.1 (0000000000000000) tentacle (stable)": 3
+	}
+}`
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
 			if args[0] == "status" {
 				return `{"fsid":"c47cac40-9bee-4d52-823b-ccd803ba5bfe","health":{"checks":{},"status":"HEALTH_ERR"},"pgmap":{"num_pgs":100,"pgs_by_state":[{"state_name":"active+clean","count":100}]}}`, nil
 			}
+			if args[0] == "versions" {
+				return cephDaemonVersions, nil
+			}
 
 			return "", nil
 		},
@@ -203,6 +214,9 @@ func TestFilesystemSubvolumeGroupController(t *testing.T) {
 				} else if args[0] == "fs" && args[1] == "subvolumegroup" && args[2] == "pin" {
 					return "", nil
 				}
+				if args[0] == "versions" {
+					return cephDaemonVersions, nil
+				}
 
 				return "", errors.Errorf("unknown command. %v", args)
 			},
diff --git a/pkg/operator/ceph/pool/radosnamespace/controller.go b/pkg/operator/ceph/pool/radosnamespace/controller.go
index 7f855c9e7..e73aa81d4 100644
--- a/pkg/operator/ceph/pool/radosnamespace/controller.go
+++ b/pkg/operator/ceph/pool/radosnamespace/controller.go
@@ -28,6 +28,7 @@ import (
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/rook/rook/pkg/operator/ceph/config"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/csi"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
@@ -224,6 +225,13 @@ func (r *ReconcileCephBlockPoolRadosNamespace) reconcile(request reconcile.Reque
 	}
 	r.clusterInfo.Context = r.opManagerContext
 
+	// Detect running Ceph version
+	runningCephVersion, err := cephclient.LeastUptodateDaemonVersion(r.context, r.clusterInfo, config.OsdType)
+	if err != nil {
+		return reconcile.Result{}, radosNamespace, errors.Wrapf(err, "failed to retrieve current ceph %q version", config.OsdType)
+	}
+	r.clusterInfo.CephVersion = runningCephVersion
+
 	// DELETE: the CR was deleted
 	if !radosNamespace.GetDeletionTimestamp().IsZero() {
 		cephRNSList := &cephv1.CephBlockPoolRadosNamespaceList{}
@@ -380,7 +388,7 @@ func (r *ReconcileCephBlockPoolRadosNamespace) updateClusterConfig(cephBlockPool
 				FuseMountOptions:   r.clusterInfo.CSIDriverSpec.CephFS.FuseMountOptions,
 			},
 			ReadAffinity: cephcsi.ReadAffinity{
-				Enabled:             r.clusterInfo.CSIDriverSpec.ReadAffinity.Enabled,
+				Enabled:             csi.ReadAffinityEnabled(r.clusterInfo.CSIDriverSpec.ReadAffinity.Enabled, r.clusterInfo.CephVersion),
 				CrushLocationLabels: r.clusterInfo.CSIDriverSpec.ReadAffinity.CrushLocationLabels,
 			},
 		},
diff --git a/pkg/operator/ceph/pool/radosnamespace/controller_test.go b/pkg/operator/ceph/pool/radosnamespace/controller_test.go
index 66049a815..946f6b197 100644
--- a/pkg/operator/ceph/pool/radosnamespace/controller_test.go
+++ b/pkg/operator/ceph/pool/radosnamespace/controller_test.go
@@ -77,11 +77,23 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 		cephBlockPoolRadosNamespace,
 	}
 
+	cephDaemonVersions := `{
+	"mon": {
+		"ceph version 20.2.1 (0000000000000000) tentacle (stable)": 3
+	},
+	"osd": {
+		"ceph version 20.2.1 (0000000000000000) tentacle (stable)": 3
+	}
+}`
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			logger.Infof("COMMAND: %s %v", command, args)
 			if args[0] == "status" {
 				return `{"fsid":"c47cac40-9bee-4d52-823b-ccd803ba5bfe","health":{"checks":{},"status":"HEALTH_ERR"},"pgmap":{"num_pgs":100,"pgs_by_state":[{"state_name":"active+clean","count":100}]}}`, nil
 			}
+			if args[0] == "versions" {
+				return cephDaemonVersions, nil
+			}
 
 			return "", nil
 		},
@@ -220,6 +232,9 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 				if args[0] == "mirror" && args[1] == "pool" {
 					return `{"mode":"disabled"}`, nil
 				}
+				if args[0] == "versions" {
+					return cephDaemonVersions, nil
+				}
 
 				return "", nil
 			},
@@ -341,6 +356,9 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 				if args[0] == "mirror" && args[1] == "pool" {
 					return `{"mode":""}`, nil
 				}
+				if args[0] == "versions" {
+					return cephDaemonVersions, nil
+				}
 
 				return "", nil
 			},
@@ -388,6 +406,9 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 				if args[0] == "mirror" && args[1] == "pool" {
 					return `{"mode":""}`, nil
 				}
+				if args[0] == "versions" {
+					return cephDaemonVersions, nil
+				}
 
 				return "", nil
 			},
@@ -440,6 +461,9 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 				if args[0] == "mirror" && args[1] == "pool" && args[2] == "info" {
 					return `{}`, nil
 				}
+				if args[0] == "versions" {
+					return cephDaemonVersions, nil
+				}
 				return "", nil
 			},
 		}
@@ -491,6 +515,9 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 				if args[0] == "mirror" && args[1] == "pool" && args[2] == "info" {
 					return `{}`, nil
 				}
+				if args[0] == "versions" {
+					return cephDaemonVersions, nil
+				}
 				return "", nil
 			},
 		}
@@ -544,6 +571,9 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 				if args[0] == "mirror" && args[1] == "pool" && args[2] == "info" {
 					return `{}`, nil
 				}
+				if args[0] == "versions" {
+					return cephDaemonVersions, nil
+				}
 				return "", nil
 			},
 		}
@@ -601,6 +631,9 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 					assert.Equal(t, cephBlockPool.Name+"/"+cephBlockPoolRadosNamespace.Name, args[3])
 					return `{}`, nil
 				}
+				if args[0] == "versions" {
+					return cephDaemonVersions, nil
+				}
 				return "", nil
 			},
 		}