diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go
index 2a10f93b1..ea6ec34a7 100644
--- a/pkg/operator/ceph/cluster/controller.go
+++ b/pkg/operator/ceph/cluster/controller.go
@@ -58,6 +58,7 @@ import (
 const (
 	crushConfigMapName       = "rook-crush-config"
 	crushmapCreatedKey       = "initialCrushMapCreated"
+	enableFlexDriver         = "ROOK_ENABLE_FLEX_DRIVER"
 	clusterCreateInterval    = 6 * time.Second
 	clusterCreateTimeout     = 60 * time.Minute
 	updateClusterInterval    = 30 * time.Second
@@ -759,8 +760,15 @@ func (c *ClusterController) onDelete(obj interface{}) {
 }
 
 func (c *ClusterController) handleDelete(cluster *cephv1.CephCluster, retryInterval time.Duration) error {
-
+	flexDriverEnabled := os.Getenv(enableFlexDriver) != "false"
 	operatorNamespace := os.Getenv(k8sutil.PodNamespaceEnvVar)
+
+	if !flexDriverEnabled {
+		logger.Debugf("Flex driver disabled: no volume attachments for cluster %s (operator namespace: %s)",
+			cluster.Namespace, operatorNamespace)
+		return nil
+	}
+
 	retryCount := 0
 	for {
 		// TODO: filter this List operation by cluster namespace on the server side
diff --git a/pkg/operator/ceph/cluster/controller_test.go b/pkg/operator/ceph/cluster/controller_test.go
index 2fc9f26bd..37e83e4e8 100644
--- a/pkg/operator/ceph/cluster/controller_test.go
+++ b/pkg/operator/ceph/cluster/controller_test.go
@@ -34,13 +34,15 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-func TestClusterDelete(t *testing.T) {
+func TestClusterDeleteFlexEnabled(t *testing.T) {
 	nodeName := "node841"
 	clusterName := "cluster684"
 	pvName := "pvc-540"
 	rookSystemNamespace := "rook-system-6413"
 
+	os.Setenv("ROOK_ENABLE_FLEX_DRIVER", "true")
 	os.Setenv(k8sutil.PodNamespaceEnvVar, rookSystemNamespace)
+	defer os.Unsetenv("ROOK_ENABLE_FLEX_DRIVER")
 	defer os.Unsetenv(k8sutil.PodNamespaceEnvVar)
 
 	clientset := testop.New(3)
@@ -95,6 +97,47 @@ func TestClusterDelete(t *testing.T) {
 	assert.Equal(t, 2, listCount)
 }
 
+func TestClusterDeleteFlexDisabled(t *testing.T) {
+	clusterName := "cluster684"
+	rookSystemNamespace := "rook-system-6413"
+
+	os.Setenv("ROOK_ENABLE_FLEX_DRIVER", "false")
+	os.Setenv(k8sutil.PodNamespaceEnvVar, rookSystemNamespace)
+	defer os.Unsetenv("ROOK_ENABLE_FLEX_DRIVER")
+	defer os.Unsetenv(k8sutil.PodNamespaceEnvVar)
+
+	clientset := testop.New(3)
+	context := &clusterd.Context{
+		Clientset: clientset,
+	}
+
+	listCount := 0
+	volumeAttachmentController := &attachment.MockAttachment{
+		MockList: func(namespace string) (*rookalpha.VolumeList, error) {
+			listCount++
+			return &rookalpha.VolumeList{Items: []rookalpha.Volume{}}, nil
+
+		},
+	}
+	callbacks := []func(clusterSpec *cephv1.ClusterSpec) error{
+		func(clusterSpec *cephv1.ClusterSpec) error {
+			logger.Infof("test success callback")
+			os.Setenv("ROOK_ENABLE_FLEX_DRIVER", "true")
+			os.Setenv(k8sutil.PodNamespaceEnvVar, rookSystemNamespace)
+			defer os.Unsetenv("ROOK_ENABLE_FLEX_DRIVER")
+			return nil
+		},
+	}
+
+	// create the cluster controller and tell it that the cluster has been deleted
+	controller := NewClusterController(context, "", volumeAttachmentController, callbacks)
+	clusterToDelete := &cephv1.CephCluster{ObjectMeta: metav1.ObjectMeta{Namespace: clusterName}}
+	controller.handleDelete(clusterToDelete, time.Microsecond)
+
+	// Ensure that the listing of volume attachments was never called.
+	assert.Equal(t, 0, listCount)
+}
+
 func TestClusterChanged(t *testing.T) {
 	// a new node added, should be a change
 	old := cephv1.ClusterSpec{