diff --git a/Documentation/Storage-Configuration/ceph-teardown.md b/Documentation/Storage-Configuration/ceph-teardown.md
index 4b8917ecdff4..20ab5c0cafce 100644
--- a/Documentation/Storage-Configuration/ceph-teardown.md
+++ b/Documentation/Storage-Configuration/ceph-teardown.md
@@ -200,6 +200,7 @@ Once the cleanup job is completed successfully, Rook will remove the finalizers
 
 This cleanup is supported only for the following custom resources:
 
-| Custom Resource                                | Ceph Resources to be cleaned up |
-| --------                                       | ------- |
-| CephFilesystemSubVolumeGroup                   | CSI stored RADOS OMAP details for pvc/volumesnapshots, subvolume snapshots, subvolume clones, subvolumes |
+| Custom Resource                      | Ceph Resources to be cleaned up |
+| --------                             | ------- |
+| CephFilesystemSubVolumeGroup         | CSI stored RADOS OMAP details for pvc/volumesnapshots, subvolume snapshots, subvolume clones, subvolumes |
+| CephBlockPoolRadosNamespace          | Images and snapshots in the RADOS namespace|
diff --git a/cmd/rook/ceph/cleanup.go b/cmd/rook/ceph/cleanup.go
index de3bf72838de..5294a24cf60c 100644
--- a/cmd/rook/ceph/cleanup.go
+++ b/cmd/rook/ceph/cleanup.go
@@ -56,6 +56,12 @@ var cleanUpSubVolumeGroupCmd = &cobra.Command{
 	Short: "Starts the cleanup process of a CephFilesystemSubVolumeGroup",
 }
 
+var cleanUpRadosNamespaceCmd = &cobra.Command{
+	// the subcommand matches CRD kind of the custom resource to be cleaned up
+	Use:   "CephBlockPoolRadosNamespace",
+	Short: "Starts the cleanup process for a CephBlockPoolRadosNamespace",
+}
+
 func init() {
 	cleanUpCmd.Flags().StringVar(&dataDirHostPath, "data-dir-host-path", "", "dataDirHostPath on the node")
 	cleanUpCmd.Flags().StringVar(&namespaceDir, "namespace-dir", "", "dataDirHostPath on the node")
@@ -68,10 +74,11 @@ func init() {
 
 	flags.SetFlagsFromEnv(cleanUpSubVolumeGroupCmd.Flags(), rook.RookEnvVarPrefix)
 
-	cleanUpCmd.AddCommand(cleanUpHostCmd, cleanUpSubVolumeGroupCmd)
+	cleanUpCmd.AddCommand(cleanUpHostCmd, cleanUpSubVolumeGroupCmd, cleanUpRadosNamespaceCmd)
 
 	cleanUpHostCmd.RunE = startHostCleanUp
 	cleanUpSubVolumeGroupCmd.RunE = startSubVolumeGroupCleanUp
+	cleanUpRadosNamespaceCmd.RunE = startRadosNamespaceCleanup
 }
 
 func startHostCleanUp(cmd *cobra.Command, args []string) error {
@@ -140,3 +147,30 @@ func startSubVolumeGroupCleanUp(cmd *cobra.Command, args []string) error {
 
 	return nil
 }
+
+func startRadosNamespaceCleanup(cmd *cobra.Command, args []string) error {
+	rook.SetLogLevel()
+	rook.LogStartupInfo(cleanUpRadosNamespaceCmd.Flags())
+
+	ctx := cmd.Context()
+	context := createContext()
+	namespace := os.Getenv(k8sutil.PodNamespaceEnvVar)
+	clusterInfo := client.AdminClusterInfo(ctx, namespace, "")
+
+	poolName := os.Getenv(opcontroller.CephBlockPoolNameEnv)
+	if poolName == "" {
+		rook.TerminateFatal(fmt.Errorf("cephblockpool name is not available in the pod environment variables"))
+	}
+
+	radosNamespace := os.Getenv(opcontroller.CephBlockPoolRadosNamespaceEnv)
+	if radosNamespace == "" {
+		rook.TerminateFatal(fmt.Errorf("cephblockpool radosNamespace is not available in the pod environment variables"))
+	}
+
+	err := cleanup.RadosNamespaceCleanup(context, clusterInfo, poolName, radosNamespace)
+	if err != nil {
+		rook.TerminateFatal(fmt.Errorf("failed to cleanup cephBlockPoolRadosNamespace %q resources in the pool %q. %v", radosNamespace, poolName, err))
+	}
+
+	return nil
+}
diff --git a/pkg/daemon/ceph/cleanup/radosnamespace.go b/pkg/daemon/ceph/cleanup/radosnamespace.go
new file mode 100644
index 000000000000..1ebb6f09a761
--- /dev/null
+++ b/pkg/daemon/ceph/cleanup/radosnamespace.go
@@ -0,0 +1,71 @@
+/*
+Copyright 2024 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cleanup
+
+import (
+	"github.com/pkg/errors"
+	"github.com/rook/rook/pkg/clusterd"
+	"github.com/rook/rook/pkg/daemon/ceph/client"
+	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
+)
+
+func RadosNamespaceCleanup(context *clusterd.Context, clusterInfo *client.ClusterInfo, poolName, radosNamespace string) error {
+	logger.Infof("starting clean up of CephBlockPoolRadosNamespace %q resources in cephblockpool %q", radosNamespace, poolName)
+
+	images, err := cephclient.ListImagesInRadosNamespace(context, clusterInfo, poolName, radosNamespace)
+	if err != nil {
+		return errors.Wrapf(err, "failed to list images in cephblockpool %q in rados namespace %q", poolName, radosNamespace)
+	}
+
+	var retErr error
+	for _, image := range images {
+		snaps, err := cephclient.ListSnapshotsInRadosNamespace(context, clusterInfo, poolName, image.Name, radosNamespace)
+		if err != nil {
+			retErr = errors.Wrapf(err, "failed to list snapshots for the image %q in cephblockpool %q in rados namespace %q.", image.Name, poolName, radosNamespace)
+			logger.Error(retErr)
+		}
+
+		for _, snap := range snaps {
+			err := cephclient.DeleteSnapshotInRadosNamespace(context, clusterInfo, poolName, image.Name, snap.Name, radosNamespace)
+			if err != nil {
+				retErr = errors.Wrapf(err, "failed to delete snapshot %q of the image %q in cephblockpool %q in rados namespace %q.", snap.Name, image.Name, poolName, radosNamespace)
+				logger.Error(retErr)
+			} else {
+				logger.Infof("successfully deleted snapshot %q of image %q in cephblockpool %q in rados namespace %q", snap.Name, image.Name, poolName, radosNamespace)
+			}
+		}
+
+		err = cephclient.MoveImageToTrashInRadosNamespace(context, clusterInfo, poolName, image.Name, radosNamespace)
+		if err != nil {
+			retErr = errors.Wrapf(err, "failed to move image %q to trash in cephblockpool %q in rados namespace %q.", image.Name, poolName, radosNamespace)
+			logger.Error(retErr)
+		}
+		err = cephclient.DeleteImageFromTrashInRadosNamespace(context, clusterInfo, poolName, image.ID, radosNamespace)
+		if err != nil {
+			retErr = errors.Wrapf(err, "failed to add task to remove image %q from trash in cephblockpool %q in rados namespace %q.", image.Name, poolName, radosNamespace)
+			logger.Error(retErr)
+		}
+	}
+
+	if retErr != nil {
+		logger.Errorf("failed to clean up CephBlockPoolRadosNamespace %q resources in cephblockpool %q", radosNamespace, poolName)
+		return retErr
+	}
+
+	logger.Infof("successfully cleaned up CephBlockPoolRadosNamespace %q resources in cephblockpool %q", radosNamespace, poolName)
+	return nil
+}
diff --git a/pkg/daemon/ceph/cleanup/radosnamespace_test.go b/pkg/daemon/ceph/cleanup/radosnamespace_test.go
new file mode 100644
index 000000000000..eb6d7aca2cb0
--- /dev/null
+++ b/pkg/daemon/ceph/cleanup/radosnamespace_test.go
@@ -0,0 +1,92 @@
+/*
+Copyright 2024 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cleanup
+
+import (
+	"testing"
+
+	"github.com/pkg/errors"
+	"github.com/rook/rook/pkg/clusterd"
+	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
+	exectest "github.com/rook/rook/pkg/util/exec/test"
+	"github.com/stretchr/testify/assert"
+)
+
+const (
+	mockImageLSResponse   = `[{"image":"csi-vol-136268e8-5386-4453-a6bd-9dca381d187d","id":"16e35cfa56a7","size":1073741824,"format":2}]`
+	mockSnapshotsResponse = `[{"id":5,"name":"snap1","size":1073741824,"protected":"false","timestamp":"Fri Apr 12 13:39:28 2024"}]`
+)
+
+func TestRadosNamespace(t *testing.T) {
+	clusterInfo := cephclient.AdminTestClusterInfo("mycluster")
+	poolName := "test-pool"
+	radosNamespace := "test-namespace"
+
+	t.Run("no images in rados namespace", func(t *testing.T) {
+		executor := &exectest.MockExecutor{}
+		executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
+			logger.Infof("Command: %s %v", command, args)
+			if args[0] == "ls" && args[1] == "-l" {
+				assert.Equal(t, poolName, args[2])
+				return "", nil
+			}
+			return "", errors.New("unknown command")
+		}
+		context := &clusterd.Context{Executor: executor}
+		err := RadosNamespaceCleanup(context, clusterInfo, poolName, radosNamespace)
+		assert.NoError(t, err)
+	})
+
+	t.Run("images with snapshots available in rados namespace", func(t *testing.T) {
+		executor := &exectest.MockExecutor{}
+		executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
+			logger.Infof("Command: %s %v", command, args)
+			// list all subvolumes in subvolumegroup
+			if args[0] == "ls" && args[1] == "-l" {
+				assert.Equal(t, poolName, args[2])
+				return mockImageLSResponse, nil
+			}
+			if args[0] == "snap" && args[1] == "ls" {
+				assert.Equal(t, "test-pool/csi-vol-136268e8-5386-4453-a6bd-9dca381d187d", args[2])
+				assert.Equal(t, "--namespace", args[3])
+				assert.Equal(t, radosNamespace, args[4])
+				return mockSnapshotsResponse, nil
+			}
+			if args[0] == "snap" && args[1] == "rm" {
+				assert.Equal(t, "test-pool/csi-vol-136268e8-5386-4453-a6bd-9dca381d187d@snap1", args[2])
+				assert.Equal(t, "--namespace", args[3])
+				assert.Equal(t, radosNamespace, args[4])
+				return "", nil
+			}
+			if args[0] == "trash" && args[1] == "mv" {
+				assert.Equal(t, "test-pool/csi-vol-136268e8-5386-4453-a6bd-9dca381d187d", args[2])
+				assert.Equal(t, "--namespace", args[3])
+				assert.Equal(t, radosNamespace, args[4])
+				return "", nil
+			}
+			if args[0] == "rbd" && args[1] == "task" && args[2] == "add" && args[3] == "trash" {
+				// pool-name/rados-namespace/image-id
+				assert.Equal(t, "test-pool/test-namespace/16e35cfa56a7", args[5])
+				return "", nil
+			}
+			return "", errors.New("unknown command")
+		}
+		context := &clusterd.Context{Executor: executor}
+		err := RadosNamespaceCleanup(context, clusterInfo, poolName, radosNamespace)
+		assert.NoError(t, err)
+	})
+}
diff --git a/pkg/daemon/ceph/client/image.go b/pkg/daemon/ceph/client/image.go
index c404d7b16ce0..f16fa2334277 100644
--- a/pkg/daemon/ceph/client/image.go
+++ b/pkg/daemon/ceph/client/image.go
@@ -35,14 +35,28 @@ const (
 )
 
 type CephBlockImage struct {
+	ID       string `json:"id"`
 	Name     string `json:"image"`
 	Size     uint64 `json:"size"`
 	Format   int    `json:"format"`
 	InfoName string `json:"name"`
 }
 
-func ListImages(context *clusterd.Context, clusterInfo *ClusterInfo, poolName string) ([]CephBlockImage, error) {
+type CephBlockImageSnapshot struct {
+	Name string `json:"name"`
+}
+
+// ListImagesInPool returns a list of images created in a cephblockpool
+func ListImagesInPool(context *clusterd.Context, clusterInfo *ClusterInfo, poolName string) ([]CephBlockImage, error) {
+	return ListImagesInRadosNamespace(context, clusterInfo, poolName, "")
+}
+
+// ListImagesInRadosNamespace returns a list if images created in a cephblockpool rados namespace
+func ListImagesInRadosNamespace(context *clusterd.Context, clusterInfo *ClusterInfo, poolName, namespace string) ([]CephBlockImage, error) {
 	args := []string{"ls", "-l", poolName}
+	if namespace != "" {
+		args = append(args, "--namespace", namespace)
+	}
 	cmd := NewRBDCommand(context, clusterInfo, args)
 	cmd.JsonOutput = true
 	buf, err := cmd.Run()
@@ -68,6 +82,56 @@ func ListImages(context *clusterd.Context, clusterInfo *ClusterInfo, poolName st
 	return images, nil
 }
 
+// ListSnapshotsInRadosNamespace lists all the snapshots created for an image in a cephblockpool in a given rados namespace
+func ListSnapshotsInRadosNamespace(context *clusterd.Context, clusterInfo *ClusterInfo, poolName, imageName, namespace string) ([]CephBlockImageSnapshot, error) {
+	snapshots := []CephBlockImageSnapshot{}
+	args := []string{"snap", "ls", getImageSpec(imageName, poolName), "--namespace", namespace}
+	cmd := NewRBDCommand(context, clusterInfo, args)
+	cmd.JsonOutput = true
+	buf, err := cmd.Run()
+	if err != nil {
+		return snapshots, errors.Wrapf(err, "failed to list snapshots of image %q in cephblockpool %q", imageName, poolName)
+	}
+
+	if err = json.Unmarshal(buf, &snapshots); err != nil {
+		return snapshots, errors.Wrapf(err, "unmarshal failed, raw buffer response: %s", string(buf))
+	}
+	return snapshots, nil
+}
+
+// DeleteSnapshotInRadosNamespace deletes a image snapshot created in block pool in a given rados namespace
+func DeleteSnapshotInRadosNamespace(context *clusterd.Context, clusterInfo *ClusterInfo, poolName, imageName, snapshot, namespace string) error {
+	args := []string{"snap", "rm", getImageSnapshotSpec(poolName, imageName, snapshot), "--namespace", namespace}
+	cmd := NewRBDCommand(context, clusterInfo, args)
+	_, err := cmd.Run()
+	if err != nil {
+		return errors.Wrapf(err, "failed to delete snapshot %q of image %q in cephblockpool %q", snapshot, imageName, poolName)
+	}
+	return nil
+}
+
+// MoveImageToTrashInRadosNamespace moves the cephblockpool image to trash in the rados namespace
+func MoveImageToTrashInRadosNamespace(context *clusterd.Context, clusterInfo *ClusterInfo, poolName, imageName, namespace string) error {
+	args := []string{"trash", "mv", getImageSpec(imageName, poolName), "--namespace", namespace}
+	cmd := NewRBDCommand(context, clusterInfo, args)
+	_, err := cmd.Run()
+	if err != nil {
+		return errors.Wrapf(err, "failed to move image %q in cephblockpool %q to trash", imageName, poolName)
+	}
+	return nil
+}
+
+// DeleteImageFromTrashInRadosNamespace deletes the image from trash in the rados namespace
+func DeleteImageFromTrashInRadosNamespace(context *clusterd.Context, clusterInfo *ClusterInfo, poolName, imageID, namespace string) error {
+	args := []string{"rbd", "task", "add", "trash", "remove", getImageSpecInRadosNamespace(poolName, namespace, imageID)}
+	cmd := NewCephCommand(context, clusterInfo, args)
+	_, err := cmd.Run()
+	if err != nil {
+		return errors.Wrapf(err, "failed to delete image %q in cephblockpool %q from trash", imageID, poolName)
+	}
+	return nil
+}
+
 // CreateImage creates a block storage image.
 // If dataPoolName is not empty, the image will use poolName as the metadata pool and the dataPoolname for data.
 // If size is zero an empty image will be created. Otherwise, an image will be
@@ -116,16 +180,22 @@ func CreateImage(context *clusterd.Context, clusterInfo *ClusterInfo, name, pool
 	return &CephBlockImage{Name: name, Size: newSizeBytes}, nil
 }
 
-func DeleteImage(context *clusterd.Context, clusterInfo *ClusterInfo, name, poolName string) error {
+func DeleteImageInPool(context *clusterd.Context, clusterInfo *ClusterInfo, name, poolName string) error {
+	return DeleteImageInRadosNamespace(context, clusterInfo, name, poolName, "")
+}
+
+func DeleteImageInRadosNamespace(context *clusterd.Context, clusterInfo *ClusterInfo, name, poolName, namespace string) error {
 	logger.Infof("deleting rbd image %q from pool %q", name, poolName)
 	imageSpec := getImageSpec(name, poolName)
 	args := []string{"rm", imageSpec}
+	if namespace != "" {
+		args = append(args, "--namespace", namespace)
+	}
 	buf, err := NewRBDCommand(context, clusterInfo, args).Run()
 	if err != nil {
 		return errors.Wrapf(err, "failed to delete image %s in pool %s, output: %s",
 			name, poolName, string(buf))
 	}
-
 	return nil
 }
 
@@ -196,3 +266,10 @@ func UnMapImage(context *clusterd.Context, clusterInfo *ClusterInfo, imageName,
 func getImageSpec(name, poolName string) string {
 	return fmt.Sprintf("%s/%s", poolName, name)
 }
+
+func getImageSpecInRadosNamespace(poolName, namespace, imageID string) string {
+	return fmt.Sprintf("%s/%s/%s", poolName, namespace, imageID)
+}
+func getImageSnapshotSpec(poolName, imageName, snapshot string) string {
+	return fmt.Sprintf("%s/%s@%s", poolName, imageName, snapshot)
+}
diff --git a/pkg/daemon/ceph/client/image_test.go b/pkg/daemon/ceph/client/image_test.go
index 335829d95dea..36710641c05b 100644
--- a/pkg/daemon/ceph/client/image_test.go
+++ b/pkg/daemon/ceph/client/image_test.go
@@ -187,7 +187,7 @@ func TestListImageLogLevelInfo(t *testing.T) {
 	}
 
 	clusterInfo := AdminTestClusterInfo("mycluster")
-	images, err = ListImages(context, clusterInfo, "pool1")
+	images, err = ListImagesInPool(context, clusterInfo, "pool1")
 	assert.Nil(t, err)
 	assert.NotNil(t, images)
 	assert.True(t, len(images) == 3)
@@ -195,7 +195,7 @@ func TestListImageLogLevelInfo(t *testing.T) {
 	listCalled = false
 
 	emptyListResult = true
-	images, err = ListImages(context, clusterInfo, "pool1")
+	images, err = ListImagesInPool(context, clusterInfo, "pool1")
 	assert.Nil(t, err)
 	assert.NotNil(t, images)
 	assert.True(t, len(images) == 0)
@@ -251,7 +251,7 @@ func TestListImageLogLevelDebug(t *testing.T) {
 	}
 
 	clusterInfo := AdminTestClusterInfo("mycluster")
-	images, err = ListImages(context, clusterInfo, "pool1")
+	images, err = ListImagesInPool(context, clusterInfo, "pool1")
 	assert.Nil(t, err)
 	assert.NotNil(t, images)
 	assert.True(t, len(images) == 3)
@@ -259,7 +259,7 @@ func TestListImageLogLevelDebug(t *testing.T) {
 	listCalled = false
 
 	emptyListResult = true
-	images, err = ListImages(context, clusterInfo, "pool1")
+	images, err = ListImagesInPool(context, clusterInfo, "pool1")
 	assert.Nil(t, err)
 	assert.NotNil(t, images)
 	assert.True(t, len(images) == 0)
diff --git a/pkg/operator/ceph/cluster/osd/osd.go b/pkg/operator/ceph/cluster/osd/osd.go
index 812cc5e4fe70..1c9b2b5c4839 100644
--- a/pkg/operator/ceph/cluster/osd/osd.go
+++ b/pkg/operator/ceph/cluster/osd/osd.go
@@ -173,12 +173,7 @@ func (osdProps osdProperties) getPreparePlacement() cephv1.Placement {
 	return osdProps.placement
 }
 
-// Start the osd management
-func (c *Cluster) Start() error {
-	namespace := c.clusterInfo.Namespace
-	config := c.newProvisionConfig()
-	errs := newProvisionErrors()
-
+func (c *Cluster) validateOSDSettings() error {
 	// Validate pod's memory if specified
 	for resourceKey, resourceValue := range c.spec.Resources {
 		if strings.HasPrefix(resourceKey, cephv1.ResourcesKeyOSD) {
@@ -188,6 +183,25 @@ func (c *Cluster) Start() error {
 			}
 		}
 	}
+	deviceSetNames := map[string]bool{}
+	for _, deviceSet := range c.spec.Storage.StorageClassDeviceSets {
+		if deviceSetNames[deviceSet.Name] {
+			return errors.Errorf("device set %q name is duplicated, OSDs cannot be configured", deviceSet.Name)
+		}
+		deviceSetNames[deviceSet.Name] = true
+	}
+	return nil
+}
+
+// Start the osd management
+func (c *Cluster) Start() error {
+	namespace := c.clusterInfo.Namespace
+	config := c.newProvisionConfig()
+	errs := newProvisionErrors()
+
+	if err := c.validateOSDSettings(); err != nil {
+		return err
+	}
 	logger.Infof("start running osds in namespace %q", namespace)
 
 	if !c.spec.Storage.UseAllNodes && len(c.spec.Storage.Nodes) == 0 && len(c.spec.Storage.StorageClassDeviceSets) == 0 {
diff --git a/pkg/operator/ceph/cluster/osd/osd_test.go b/pkg/operator/ceph/cluster/osd/osd_test.go
index 2ee0d7567ce8..043e41c7f53d 100644
--- a/pkg/operator/ceph/cluster/osd/osd_test.go
+++ b/pkg/operator/ceph/cluster/osd/osd_test.go
@@ -984,3 +984,36 @@ func TestGetOSDLocationFromArgs(t *testing.T) {
 	assert.Equal(t, osdLocaiton, "")
 	assert.Equal(t, locationFound, false)
 }
+
+func TestValidateOSDSettings(t *testing.T) {
+	namespace := "ns"
+	clusterInfo := &cephclient.ClusterInfo{
+		Namespace:   namespace,
+		CephVersion: cephver.Quincy,
+		Context:     context.TODO(),
+	}
+	clusterInfo.SetName("rook-ceph-test")
+	c := New(&clusterd.Context{}, clusterInfo, cephv1.ClusterSpec{}, "version")
+
+	t.Run("validate with no settings", func(t *testing.T) {
+		assert.NoError(t, c.validateOSDSettings())
+	})
+
+	t.Run("valid device sets", func(t *testing.T) {
+		c.spec.Storage.StorageClassDeviceSets = []cephv1.StorageClassDeviceSet{
+			{Name: "set1"},
+			{Name: "set2"},
+			{Name: "set3"},
+		}
+		assert.NoError(t, c.validateOSDSettings())
+	})
+
+	t.Run("duplicate device sets", func(t *testing.T) {
+		c.spec.Storage.StorageClassDeviceSets = []cephv1.StorageClassDeviceSet{
+			{Name: "set1"},
+			{Name: "set2"},
+			{Name: "set1"},
+		}
+		assert.Error(t, c.validateOSDSettings())
+	})
+}
diff --git a/pkg/operator/ceph/cluster/osd/spec.go b/pkg/operator/ceph/cluster/osd/spec.go
index 7f96c8ea2120..96a9570cf33f 100644
--- a/pkg/operator/ceph/cluster/osd/spec.go
+++ b/pkg/operator/ceph/cluster/osd/spec.go
@@ -585,7 +585,7 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo, provisionC
 	} else {
 		// Add the volume to the spec and the mount to the daemon container
 		// so that it can pick the already mounted/activated osd metadata path
-		// This container will activate the OSD and place the activated filesinto an empty dir
+		// This container will activate the OSD and place the activated files into an empty dir
 		// The empty dir will be shared by the "activate-osd" pod and the "osd" main pod
 		activateOSDVolume, activateOSDContainer := c.getActivateOSDInitContainer(c.spec.DataDirHostPath, c.clusterInfo.Namespace, osdID, osd, osdProps)
 		volumes = append(volumes, activateOSDVolume...)
diff --git a/pkg/operator/ceph/controller/cleanup.go b/pkg/operator/ceph/controller/cleanup.go
index 72234271901f..01c4b23047d6 100644
--- a/pkg/operator/ceph/controller/cleanup.go
+++ b/pkg/operator/ceph/controller/cleanup.go
@@ -40,6 +40,10 @@ const (
 	CephFSNameEnv               = "FILESYSTEM_NAME"
 	CSICephFSRadosNamesaceEnv   = "CSI_CEPHFS_RADOS_NAMESPACE"
 	CephFSMetaDataPoolNameEnv   = "METADATA_POOL_NAME"
+
+	// cephblockpoolradosnamespace env resources
+	CephBlockPoolNameEnv           = "BLOCKPOOL_NAME"
+	CephBlockPoolRadosNamespaceEnv = "RADOS_NAMESPACE"
 )
 
 // ResourceCleanup defines an rook ceph resource to be cleaned up
diff --git a/pkg/operator/ceph/controller/cleanup_test.go b/pkg/operator/ceph/controller/cleanup_test.go
index 632977b71e12..bc45878686ad 100644
--- a/pkg/operator/ceph/controller/cleanup_test.go
+++ b/pkg/operator/ceph/controller/cleanup_test.go
@@ -55,10 +55,7 @@ func TestJobTemplateSpec(t *testing.T) {
 	cleanup := NewResourceCleanup(svgObj, cluster, rookImage, testConfig)
 	podTemplateSpec := cleanup.jobTemplateSpec()
 	assert.Equal(t, "CephFSSubvolumeGroup", podTemplateSpec.Spec.Containers[0].Args[2])
-	assert.Equal(t, "config1", podTemplateSpec.Spec.Containers[0].Env[3].Name)
-	assert.Equal(t, "value1", podTemplateSpec.Spec.Containers[0].Env[3].Value)
-	assert.Equal(t, "config2", podTemplateSpec.Spec.Containers[0].Env[4].Name)
-	assert.Equal(t, "value2", podTemplateSpec.Spec.Containers[0].Env[4].Value)
+	assert.Equal(t, 5, len(podTemplateSpec.Spec.Containers[0].Env))
 }
 
 func TestForceDeleteRequested(t *testing.T) {
diff --git a/pkg/operator/ceph/disruption/clusterdisruption/osd.go b/pkg/operator/ceph/disruption/clusterdisruption/osd.go
index ad82fbc32bbd..407881ed5623 100644
--- a/pkg/operator/ceph/disruption/clusterdisruption/osd.go
+++ b/pkg/operator/ceph/disruption/clusterdisruption/osd.go
@@ -223,7 +223,7 @@ func (r *ReconcileClusterDisruption) reconcilePDBsForOSDs(
 	allFailureDomains,
 	osdDownFailureDomains []string,
 	activeNodeDrains bool,
-	pgHealhtyRegex string,
+	pgHealthyRegex string,
 ) (reconcile.Result, error) {
 	var osdDown bool
 	var drainingFailureDomain string
@@ -232,7 +232,7 @@ func (r *ReconcileClusterDisruption) reconcilePDBsForOSDs(
 		drainingFailureDomain = osdDownFailureDomains[0]
 	}
 
-	pgHealthMsg, pgClean, err := cephclient.IsClusterClean(r.context.ClusterdContext, clusterInfo, pgHealhtyRegex)
+	pgHealthMsg, pgClean, err := cephclient.IsClusterClean(r.context.ClusterdContext, clusterInfo, pgHealthyRegex)
 	if err != nil {
 		// If the error contains that message, this means the cluster is not up and running
 		// No monitors are present and thus no ceph configuration has been created
diff --git a/pkg/operator/ceph/pool/radosnamespace/controller.go b/pkg/operator/ceph/pool/radosnamespace/controller.go
index 0c48e8602ca7..5e82029c3400 100644
--- a/pkg/operator/ceph/pool/radosnamespace/controller.go
+++ b/pkg/operator/ceph/pool/radosnamespace/controller.go
@@ -189,6 +189,12 @@ func (r *ReconcileCephBlockPoolRadosNamespace) reconcile(request reconcile.Reque
 					logger.Info(opcontroller.OperatorNotInitializedMessage)
 					return opcontroller.WaitForRequeueIfOperatorNotInitialized, nil
 				}
+				if opcontroller.ForceDeleteRequested(cephBlockPoolRadosNamespace.GetAnnotations()) {
+					cleanupErr := r.cleanup(cephBlockPoolRadosNamespace, &cephCluster)
+					if cleanupErr != nil {
+						return reconcile.Result{}, errors.Wrapf(cleanupErr, "failed to create clean up job for ceph blockpool rados namespace %q", namespacedName)
+					}
+				}
 				return reconcile.Result{}, errors.Wrapf(err, "failed to delete ceph blockpool rados namespace %q", cephBlockPoolRadosNamespace.Name)
 			}
 		}
@@ -356,3 +362,18 @@ func buildClusterID(cephBlockPoolRadosNamespace *cephv1.CephBlockPoolRadosNamesp
 	clusterID := fmt.Sprintf("%s-%s-block-%s", cephBlockPoolRadosNamespace.Namespace, cephBlockPoolRadosNamespace.Spec.BlockPoolName, getRadosNamespaceName(cephBlockPoolRadosNamespace))
 	return k8sutil.Hash(clusterID)
 }
+
+func (r *ReconcileCephBlockPoolRadosNamespace) cleanup(radosNamespace *cephv1.CephBlockPoolRadosNamespace, cephCluster *cephv1.CephCluster) error {
+	logger.Infof("starting cleanup of the ceph resources for radosNamesapce %q in namespace %q", radosNamespace.Name, radosNamespace.Namespace)
+	cleanupConfig := map[string]string{
+		opcontroller.CephBlockPoolNameEnv:           radosNamespace.Spec.BlockPoolName,
+		opcontroller.CephBlockPoolRadosNamespaceEnv: getRadosNamespaceName(radosNamespace),
+	}
+	cleanup := opcontroller.NewResourceCleanup(radosNamespace, cephCluster, r.opConfig.Image, cleanupConfig)
+	jobName := k8sutil.TruncateNodeNameForJob("cleanup-radosnamespace-%s", fmt.Sprintf("%s-%s", radosNamespace.Spec.BlockPoolName, radosNamespace.Name))
+	err := cleanup.StartJob(r.clusterInfo.Context, r.context.Clientset, jobName)
+	if err != nil {
+		return errors.Wrapf(err, "failed to run clean up job to clean the ceph resources in radosNamesapce %q", radosNamespace.Name)
+	}
+	return nil
+}
diff --git a/tests/framework/clients/block.go b/tests/framework/clients/block.go
index b885635f7885..55db630b5d97 100644
--- a/tests/framework/clients/block.go
+++ b/tests/framework/clients/block.go
@@ -161,7 +161,7 @@ func (b *BlockOperation) ListAllImages(clusterInfo *client.ClusterInfo) ([]Block
 func (b *BlockOperation) ListImagesInPool(clusterInfo *client.ClusterInfo, poolName string) ([]BlockImage, error) {
 	// for each pool, get further details about all the images in the pool
 	images := []BlockImage{}
-	cephImages, err := client.ListImages(b.k8sClient.MakeContext(), clusterInfo, poolName)
+	cephImages, err := client.ListImagesInPool(b.k8sClient.MakeContext(), clusterInfo, poolName)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get images from pool %s: %+v", poolName, err)
 	}
@@ -182,7 +182,7 @@ func (b *BlockOperation) ListImagesInPool(clusterInfo *client.ClusterInfo, poolN
 // DeleteBlockImage Function to list all the blocks created/being managed by rook
 func (b *BlockOperation) DeleteBlockImage(clusterInfo *client.ClusterInfo, image BlockImage) error {
 	context := b.k8sClient.MakeContext()
-	return client.DeleteImage(context, clusterInfo, image.Name, image.PoolName)
+	return client.DeleteImageInPool(context, clusterInfo, image.Name, image.PoolName)
 }
 
 // CreateClientPod starts a pod that should have a block PVC.