diff --git a/PendingReleaseNotes.md b/PendingReleaseNotes.md
index aba8912815a..d33f4348395 100644
--- a/PendingReleaseNotes.md
+++ b/PendingReleaseNotes.md
@@ -37,3 +37,12 @@
    different subvolume.
 
 ## NOTE
+
+- The RADOS lock that serializes fscrypt setup for encrypted CephFS volumes
+  is now taken in the CephFS RADOS namespace instead of the default
+  namespace of the metadata pool. This applies to every deployment with
+  encrypted volumes. `cephFS.radosNamespace` defaults to `csi`, so the lock
+  moves to this namespace even when the option was never directly
+  configured. During a rolling nodeplugin upgrade the locks in the default
+  namespace and `cephFS.radosNamespace` are taken, so that pods which have
+  not been upgraded yet stay serialized against upgraded ones.
diff --git a/e2e/cephfs.go b/e2e/cephfs.go
index d6a54e31c3f..2c17e11ca7a 100644
--- a/e2e/cephfs.go
+++ b/e2e/cephfs.go
@@ -647,6 +647,70 @@ var _ = Describe(cephfsType, func() {
 					}
 				})
 			}
+
+			It("verify the fscrypt lock is taken in the CephFS RADOS namespace", func() {
+				scOpts := map[string]string{
+					"encrypted":       "true",
+					"encryptionKMSID": "secrets-metadata-test",
+				}
+				err := createCephfsStorageClass(f.ClientSet, f, true, scOpts)
+				if err != nil {
+					logAndFail("failed to create CephFS storageclass: %v", err)
+				}
+
+				pvc, app, err := createPVCAndAppBinding(pvcPath, appPath, f, deployTimeout)
+				if err != nil {
+					logAndFail("failed to create PVC and application binding: %v", err)
+				}
+
+				imageData, err := getImageInfoFromPVC(pvc.Namespace, pvc.Name, f)
+				if err != nil {
+					logAndFail("failed to get image info from PVC: %v", err)
+				}
+
+				// Staging an encrypted volume serializes the fscrypt setup
+				// with a RADOS lock on an object named after the ObjectUUID
+				// of the volume. The lock must live in the RADOS namespace
+				// configured as cephFS.radosNamespace, where cephfsOptions()
+				// points the rados commands.
+				//
+				// The lock is released before NodeStageVolume returns, so it
+				// can not be observed directly. Instead this relies on
+				// cls_lock leaving the object behind on unlock, which holds
+				// for every lock type except LOCK_EXCLUSIVE_EPHEMERAL. If
+				// the fscrypt lock ever becomes ephemeral, this check needs
+				// a new way to observe the namespace of the lock.
+				_, stdErr, err := execCommandInToolBoxPod(f,
+					fmt.Sprintf("rados stat %s %s", imageData.imageID, cephfsOptions(metadataPool)),
+					rookNamespace)
+				if err != nil || stdErr != "" {
+					logAndFail("fscrypt lock object %s missing in the CephFS RADOS namespace: err=%v stdErr=%s",
+						imageData.imageID, err, stdErr)
+				}
+
+				// The transitional lock that serializes against nodeplugins
+				// which have not been upgraded yet is taken in the default
+				// RADOS namespace. Remove this check together with
+				// acquireLegacyEncryptionLock.
+				_, stdErr, err = execCommandInToolBoxPod(f,
+					fmt.Sprintf("rados stat %s --pool=%s", imageData.imageID, metadataPool),
+					rookNamespace)
+				if err != nil || stdErr != "" {
+					logAndFail("legacy fscrypt lock object %s missing in the default RADOS namespace: err=%v stdErr=%s",
+						imageData.imageID, err, stdErr)
+				}
+
+				err = deletePVCAndApp("", f, pvc, app)
+				if err != nil {
+					logAndFail("failed to delete PVC and application: %v", err)
+				}
+				validateOmapCount(f, 0, cephfsType, metadataPool, volumesType)
+
+				err = deleteResource(cephFSExamplePath + "storageclass.yaml")
+				if err != nil {
+					logAndFail("failed to delete CephFS storageclass: %v", err)
+				}
+			})
 		}
 
 		It("create a PVC and check PVC/PV metadata on CephFS subvolume", func() {
diff --git a/internal/cephfs/nodeserver.go b/internal/cephfs/nodeserver.go
index 846092fb897..a5af8e8dcd1 100644
--- a/internal/cephfs/nodeserver.go
+++ b/internal/cephfs/nodeserver.go
@@ -167,6 +167,15 @@ func maybeUnlockFileEncryption(
 
 	log.DebugLog(ctx, "Creating lock for the following volume ID %s", volID)
 
+	releaseLegacyLock, err := acquireLegacyEncryptionLock(ctx, volOptions, objectUUID,
+		lockName, lockCookie, lockDesc, lockDuration)
+	if err != nil {
+		log.ErrorLog(ctx, "failed to create the legacy lock for volume ID %s: %v", volID, err)
+
+		return err
+	}
+	defer releaseLegacyLock()
+
 	ioctx, err := volOptions.GetConnection().GetIoctx(volOptions.MetadataPool)
 	if err != nil {
 		log.ErrorLog(ctx, "Failed to create ioctx: %s", err)
@@ -175,6 +184,8 @@ func maybeUnlockFileEncryption(
 	}
 	defer ioctx.Destroy()
 
+	ioctx.SetNamespace(volOptions.RadosNamespace)
+
 	lock := iolock.NewLock(ioctx, objectUUID, lockName, lockCookie, lockDesc, lockDuration)
 	err = lock.LockExclusive(ctx)
 	if err != nil {
@@ -194,6 +205,57 @@ func maybeUnlockFileEncryption(
 	return nil
 }
 
+// acquireLegacyEncryptionLock takes the fscrypt lock in the default RADOS
+// namespace of the metadata pool, where releases before the lock moved to the
+// volume's RADOS namespace took it.
+//
+// The returned function releases the lock and must always be called.
+//
+// TODO: remove once upgrades from releases that lock in the default RADOS
+// namespace are no longer supported.
+func acquireLegacyEncryptionLock(
+	ctx context.Context,
+	volOptions *store.VolumeOptions,
+	objectUUID, lockName, lockCookie, lockDesc string,
+	lockDuration time.Duration,
+) (func(), error) {
+	noop := func() {}
+
+	// Skip when the volume has no RADOS namespace. In that case the caller
+	// already takes its lock in the default namespace of the pool, which is
+	// exactly where the legacy lock lives, and locking the same object twice
+	// with the same cookie fails with EEXIST.
+	if volOptions.RadosNamespace == "" {
+		return noop, nil
+	}
+
+	ioctx, err := volOptions.GetConnection().GetIoctx(volOptions.MetadataPool)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create ioctx in the default namespace of pool %q: %w",
+			volOptions.MetadataPool, err)
+	}
+
+	lock := iolock.NewLock(ioctx, objectUUID, lockName, lockCookie, lockDesc, lockDuration)
+	if err = lock.LockExclusive(ctx); err != nil {
+		ioctx.Destroy()
+
+		if errors.Is(err, iolock.ErrLockNotPermitted) {
+			log.DebugLog(ctx, "not allowed to lock in the default namespace of pool %q, "+
+				"skipping the legacy lock for object %s: %v",
+				volOptions.MetadataPool, objectUUID, err)
+
+			return noop, nil
+		}
+
+		return nil, err
+	}
+
+	return func() {
+		lock.Unlock(ctx)
+		ioctx.Destroy()
+	}, nil
+}
+
 // generateLockCookie generates a consistent lock cookie for the client.
 func generateLockCookie() string {
 	hostname, err := os.Hostname()
diff --git a/internal/util/lock/lock.go b/internal/util/lock/lock.go
index d201f10b781..26729ad69c4 100644
--- a/internal/util/lock/lock.go
+++ b/internal/util/lock/lock.go
@@ -18,6 +18,7 @@ package lock
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"syscall"
 	"time"
@@ -27,6 +28,12 @@ import (
 	"github.com/ceph/ceph-csi/internal/util/log"
 )
 
+// ErrLockNotPermitted is returned when the CephX credentials are not allowed to
+// perform the lock operation. Locking is a RADOS class operation, so it needs
+// the class-exec permission on the pool, and caps that cover the RADOS
+// namespace of the IO context.
+var ErrLockNotPermitted = errors.New("not allowed to lock")
+
 // IOCtxLock provides methods for acquiring and releasing exclusive locks on a volume.
 // using rados IO context locks.
 type IOCtxLock interface {
@@ -82,6 +89,8 @@ func (lck *lock) LockExclusive(ctx context.Context) error {
 		case -int(syscall.EEXIST):
 			return fmt.Errorf("lock is already held by the same client and cookie pair for %v volume",
 				lck.volID)
+		case -int(syscall.EPERM), -int(syscall.EACCES):
+			return fmt.Errorf("%w volume ID %v", ErrLockNotPermitted, lck.volID)
 		default:
 			return fmt.Errorf("failed to lock volume ID %v: %w", lck.volID, err)
 		}