diff --git a/pkg/syncer/cnsoperator/controller/cnsnodevmattachment/cnsnodevmattachment_controller.go b/pkg/syncer/cnsoperator/controller/cnsnodevmattachment/cnsnodevmattachment_controller.go
index eb335c350c..4030f5dca9 100644
--- a/pkg/syncer/cnsoperator/controller/cnsnodevmattachment/cnsnodevmattachment_controller.go
+++ b/pkg/syncer/cnsoperator/controller/cnsnodevmattachment/cnsnodevmattachment_controller.go
@@ -50,6 +50,8 @@ import (
 	csifault "sigs.k8s.io/vsphere-csi-driver/v3/pkg/common/fault"
 	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/common/prometheus"
 	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/common/utils"
+	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/csi/service/common"
+	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/csi/service/common/commonco"
 	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/csi/service/logger"
 	k8s "sigs.k8s.io/vsphere-csi-driver/v3/pkg/kubernetes"
 	cnsoptypes "sigs.k8s.io/vsphere-csi-driver/v3/pkg/syncer/cnsoperator/types"
@@ -69,6 +71,7 @@ const (
 var (
 	backOffDuration         map[k8stypes.NamespacedName]time.Duration
 	backOffDurationMapMutex = sync.Mutex{}
+	isSharedDiskEnabled     bool
 )
 
 // Mockable function variables for testing
@@ -135,6 +138,9 @@ func Add(mgr manager.Manager, clusterFlavor cnstypes.CnsClusterFlavor,
 		return err
 	}
 
+	// If the capability gets enabled at a later point, then container will be restarted and this value will be reinitialized.
+	isSharedDiskEnabled = commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx, common.SharedDiskFss)
+
 	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: cnsoperatorapis.GroupName})
 	return add(mgr, newReconciler(mgr, configInfo, volumeManager, vmOperatorClient, recorder))
 }
@@ -347,6 +353,17 @@ func (r *ReconcileCnsNodeVMAttachment) Reconcile(ctx context.Context,
 		}
 		nodeUUID := instance.Spec.NodeUUID
 		if !instance.Status.Attached && instance.DeletionTimestamp == nil {
+
+			if isSharedDiskEnabled {
+				// If isSharedDiskEnabled is enabled, then all new attach operations should happen via the
+				// new CnsNodeVMBatchAttachment CRD.
+				err := r.updateErrorOnInstanceToDisallowAttach(ctx, instance)
+				if err != nil {
+					return reconcile.Result{RequeueAfter: timeout}, csifault.CSIInternalFault, nil
+				}
+				return reconcile.Result{}, "", nil
+			}
+
 			nodeVM, err := getVMByUUIDFromVCenter(internalCtx, dc, nodeUUID)
 			if err != nil {
 				msg := fmt.Sprintf("failed to find the VM with UUID: %q for CnsNodeVmAttachment "+
@@ -817,6 +834,25 @@ func isVmCrPresent(ctx context.Context, vmOperatorClient client.Client,
 	return nil, nil
 }
 
+// updateErrorOnInstanceToDisallowAttach sets error on the CnsNodeVMAttachment CR
+// so that devops user can detach this volume
+// from the VM and re-attach it so that the new Batch Attach flow is triggered.
+func (r *ReconcileCnsNodeVMAttachment) updateErrorOnInstanceToDisallowAttach(ctx context.Context,
+	instance *v1a1.CnsNodeVmAttachment) error {
+	log := logger.GetLogger(ctx)
+
+	log.Infof("Attach should happen via the new CnsNodeVMBatchAttachment CRD. Skipping attach.")
+	msg := "CnsNodeVMAttachment CR is deprecated. Please detach this PVC from the VM and then reattach it."
+	instance.Status.Error = msg
+	err := k8s.UpdateStatus(ctx, r.client, instance)
+	if err != nil {
+		log.Errorf("updateCnsNodeVMAttachment failed. err: %v", err)
+		return err
+	}
+	recordEvent(ctx, r, instance, v1.EventTypeWarning, msg)
+	return nil
+}
+
 // getVCDatacenterFromConfig returns datacenter registered for each vCenter
 func getVCDatacentersFromConfig(cfg *config.Config) (map[string][]string, error) {
 	var err error
diff --git a/pkg/syncer/cnsoperator/controller/cnsnodevmattachment/cnsnodevmattachment_controller_test.go b/pkg/syncer/cnsoperator/controller/cnsnodevmattachment/cnsnodevmattachment_controller_test.go
index 0a78b59728..f6df1066e7 100644
--- a/pkg/syncer/cnsoperator/controller/cnsnodevmattachment/cnsnodevmattachment_controller_test.go
+++ b/pkg/syncer/cnsoperator/controller/cnsnodevmattachment/cnsnodevmattachment_controller_test.go
@@ -329,3 +329,60 @@ func TestReconcileDetachWithVolumeIDFallbackFailure(t *testing.T) {
 
 	t.Logf("✓ Reconcile correctly handled getVolumeID failure and requeued for retry")
 }
+
+func TestUpdateErrorOnInstanceToDisallowAttach(t *testing.T) {
+	ctx := context.Background()
+
+	// Set up the scheme
+	SchemeGroupVersion := schema.GroupVersion{
+		Group:   "cns.vmware.com",
+		Version: "v1alpha1",
+	}
+	s := scheme.Scheme
+	s.AddKnownTypes(SchemeGroupVersion, &v1a1.CnsNodeVmAttachment{})
+	metav1.AddToGroupVersion(s, SchemeGroupVersion)
+
+	// Create CnsNodeVmAttachment instance
+	instance := &v1a1.CnsNodeVmAttachment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-attachment",
+			Namespace: "test-ns",
+		},
+	}
+
+	// Create fake client with status subresource
+	fakeClient := fake.NewClientBuilder().
+		WithScheme(s).
+		WithStatusSubresource(instance).
+		WithRuntimeObjects(instance).
+		Build()
+
+	// Create reconciler
+	r := &ReconcileCnsNodeVMAttachment{
+		client:   fakeClient,
+		recorder: record.NewFakeRecorder(10),
+		scheme:   s,
+	}
+
+	backOffDuration = make(map[k8stypes.NamespacedName]time.Duration)
+
+	// Call the function
+	err := r.updateErrorOnInstanceToDisallowAttach(ctx, instance)
+
+	// Assertions
+	assert.NoError(t, err, "Function should not return an error")
+	assert.Equal(t,
+		"CnsNodeVMAttachment CR is deprecated. Please detach this PVC from the VM and then reattach it.",
+		instance.Status.Error,
+		"Status.Error should be updated with the deprecation message",
+	)
+
+	// Verify that the status was persisted by fetching the object from the fake client
+	fetched := &v1a1.CnsNodeVmAttachment{}
+	err = fakeClient.Get(ctx, k8stypes.NamespacedName{
+		Name:      "test-attachment",
+		Namespace: "test-ns",
+	}, fetched)
+	assert.NoError(t, err, "Should be able to fetch the instance from fake client")
+	assert.Equal(t, instance.Status.Error, fetched.Status.Error, "Status should be persisted in fake client")
+}