diff --git a/controllers/storagecluster/cephcluster.go b/controllers/storagecluster/cephcluster.go
index 0da359b8fe..6f09da6091 100644
--- a/controllers/storagecluster/cephcluster.go
+++ b/controllers/storagecluster/cephcluster.go
@@ -1308,8 +1308,8 @@ func isBluestore(store rookCephv1.OSDStore) bool {
 	return false
 }
 
-func (r *StorageClusterReconciler) getOsdCount(sc *ocsv1.StorageCluster) int {
-	storageClassDeviceSets := newStorageClassDeviceSets(sc, r.serverVersion)
+func getOsdCount(sc *ocsv1.StorageCluster, serverVersion *version.Info) int {
+	storageClassDeviceSets := newStorageClassDeviceSets(sc, serverVersion)
 	osdCount := 0
 	for _, ds := range storageClassDeviceSets {
 		osdCount += ds.Count
diff --git a/controllers/storagecluster/clusterclaims.go b/controllers/storagecluster/clusterclaims.go
index 79de134b9f..11d349d58f 100644
--- a/controllers/storagecluster/clusterclaims.go
+++ b/controllers/storagecluster/clusterclaims.go
@@ -9,12 +9,14 @@ import (
 	"github.com/go-logr/logr"
 	operatorsv1alpha1 "github.com/operator-framework/api/pkg/operators/v1alpha1"
 	ocsv1 "github.com/red-hat-storage/ocs-operator/v4/api/v1"
+	rookCephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	corev1 "k8s.io/api/core/v1"
 	extensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/version"
 	"k8s.io/client-go/tools/clientcmd"
 	clusterclientv1alpha1 "open-cluster-management.io/api/client/cluster/clientset/versioned"
 	clusterv1alpha1 "open-cluster-management.io/api/cluster/v1alpha1"
@@ -30,12 +32,13 @@ const (
 )
 
 var (
-	ClusterClaimGroup   = "odf"
-	OdfVersion          = fmt.Sprintf("version.%s.openshift.io", ClusterClaimGroup)
-	StorageSystemName   = fmt.Sprintf("storagesystemname.%s.openshift.io", ClusterClaimGroup)
-	StorageClusterName  = fmt.Sprintf("storageclustername.%s.openshift.io", ClusterClaimGroup)
-	StorageClusterCount = fmt.Sprintf("count.storageclusters.%s.openshift.io", ClusterClaimGroup)
-	CephFsid            = fmt.Sprintf("cephfsid.%s.openshift.io", ClusterClaimGroup)
+	ClusterClaimGroup         = "odf"
+	OdfVersion                = fmt.Sprintf("version.%s.openshift.io", ClusterClaimGroup)
+	StorageSystemName         = fmt.Sprintf("storagesystemname.%s.openshift.io", ClusterClaimGroup)
+	StorageClusterName        = fmt.Sprintf("storageclustername.%s.openshift.io", ClusterClaimGroup)
+	StorageClusterCount       = fmt.Sprintf("count.storageclusters.%s.openshift.io", ClusterClaimGroup)
+	StorageClusterDROptimized = fmt.Sprintf("droptimized.%s.openshift.io", ClusterClaimGroup)
+	CephFsid                  = fmt.Sprintf("cephfsid.%s.openshift.io", ClusterClaimGroup)
 )
 
 type ocsClusterClaim struct{}
@@ -100,11 +103,18 @@ func (obj *ocsClusterClaim) ensureCreated(r *StorageClusterReconciler, instance
 		return reconcile.Result{}, err
 	}
 
+	isDROptimized, err := creator.getIsDROptimized(r.serverVersion)
+	if err != nil {
+		r.Log.Error(err, "failed to get cephcluster status. retrying again")
+		return reconcile.Result{}, err
+	}
+
 	err = creator.setStorageClusterCount(strconv.Itoa(storageClusterCount)).
 		setStorageSystemName(storageSystemName).
 		setStorageClusterName(instance.Name).
 		setOdfVersion(odfVersion).
 		setCephFsid(cephFsid).
+		setDROptimized(isDROptimized).
 		create()
 
 	return reconcile.Result{}, err
@@ -193,6 +203,26 @@ func (c *ClusterClaimCreator) getCephFsid() (string, error) {
 	return "", fmt.Errorf("failed to fetch ceph fsid from %q secret", RookCephMonSecretName)
 }
 
+func (c *ClusterClaimCreator) getIsDROptimized(serverVersion *version.Info) (string, error) {
+	var cephCluster rookCephv1.CephCluster
+	err := c.Client.Get(c.Context, types.NamespacedName{Name: generateNameForCephClusterFromString(c.StorageCluster.Name), Namespace: c.StorageCluster.Namespace}, &cephCluster)
+	if err != nil {
+		return "false", err
+	}
+	if cephCluster.Status.CephStorage == nil || cephCluster.Status.CephStorage.OSD.StoreType == nil {
+		return "false", fmt.Errorf("cephcluster status does not have OSD store information")
+	}
+	bluestorerdr, ok := cephCluster.Status.CephStorage.OSD.StoreType["bluestore-rdr"]
+	if !ok {
+		return "false", nil
+	}
+	total := getOsdCount(c.StorageCluster, serverVersion)
+	if bluestorerdr < total {
+		return "false", nil
+	}
+	return "true", nil
+}
+
 func (c *ClusterClaimCreator) setStorageClusterCount(count string) *ClusterClaimCreator {
 	c.Values[StorageClusterCount] = count
 	return c
@@ -218,6 +248,11 @@ func (c *ClusterClaimCreator) setCephFsid(fsid string) *ClusterClaimCreator {
 	return c
 }
 
+func (c *ClusterClaimCreator) setDROptimized(optimized string) *ClusterClaimCreator {
+	c.Values[StorageClusterDROptimized] = optimized
+	return c
+}
+
 func (c *ClusterClaimCreator) getStorageSystemName() (string, error) {
 	for _, ref := range c.StorageCluster.OwnerReferences {
 		if ref.Kind == "StorageSystem" {
diff --git a/controllers/storagecluster/resourceprofile.go b/controllers/storagecluster/resourceprofile.go
index b19b862b7e..359a22af44 100644
--- a/controllers/storagecluster/resourceprofile.go
+++ b/controllers/storagecluster/resourceprofile.go
@@ -51,7 +51,7 @@ func (r *StorageClusterReconciler) ensureResourceProfileChangeApplied(sc *ocsv1.
 	}
 
 	// Verify if expected number of osd pods with the current resource profile label are running
-	if err := r.verifyDaemonWithResourceProfile(sc.Namespace, map[string]string{"app": "rook-ceph-osd", defaults.ODFResourceProfileKey: currentResourceProfile}, r.getOsdCount(sc)); err != nil {
+	if err := r.verifyDaemonWithResourceProfile(sc.Namespace, map[string]string{"app": "rook-ceph-osd", defaults.ODFResourceProfileKey: currentResourceProfile}, getOsdCount(sc, r.serverVersion)); err != nil {
 		return err
 	}