diff --git a/pkg/csi/service/common/constants.go b/pkg/csi/service/common/constants.go
index 7f905fb0ad..839dffcb42 100644
--- a/pkg/csi/service/common/constants.go
+++ b/pkg/csi/service/common/constants.go
@@ -296,6 +296,10 @@ const (
 	//AnnVolumeAccessibleTopology is the annotation set by the supervisor cluster on PVC
 	AnnVolumeAccessibleTopology = "csi.vsphere.volume-accessible-topology"
 
+	// AnnTanzuGuestClusterOwner is the annotation key to set TanzuClusterID as value
+	// on the PVC in supervisor cluster
+	AnnTanzuGuestClusterOwner = "csi.vsphere.tanzu-kubernetes-cluster"
+
 	// PVtoBackingDiskObjectIdSupportedVCenterMajor is the minimum major version of vCenter
 	// on which PV to BackingDiskObjectId mapping feature is supported.
 	PVtoBackingDiskObjectIdSupportedVCenterMajor int = 7
diff --git a/pkg/csi/service/wcpguest/controller.go b/pkg/csi/service/wcpguest/controller.go
index 136061fee6..b067c76be7 100644
--- a/pkg/csi/service/wcpguest/controller.go
+++ b/pkg/csi/service/wcpguest/controller.go
@@ -301,11 +301,11 @@ func (c *controller) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequ
 		if err != nil {
 			if errors.IsNotFound(err) {
 				diskSize := strconv.FormatInt(volSizeMB, 10) + "Mi"
-				var annotations map[string]string
+				annotations := make(map[string]string)
+				annotations[common.AnnTanzuGuestClusterOwner] = c.tanzukubernetesClusterUID
 				if !isFileVolumeRequest && commonco.ContainerOrchestratorUtility.IsFSSEnabled(ctx, common.TKGsHA) &&
 					req.AccessibilityRequirements != nil {
 					// Generate volume topology requirement annotation.
-					annotations = make(map[string]string)
 					topologyAnnotation, err := generateGuestClusterRequestedTopologyJSON(req.AccessibilityRequirements.Preferred)
 					if err != nil {
 						msg := fmt.Sprintf("failed to generate accessibility topology for pvc with name: %s "+
diff --git a/pkg/syncer/pvcsi_fullsync.go b/pkg/syncer/pvcsi_fullsync.go
index 960ed09a5f..b336bb54be 100644
--- a/pkg/syncer/pvcsi_fullsync.go
+++ b/pkg/syncer/pvcsi_fullsync.go
@@ -18,12 +18,15 @@ package syncer
 
 import (
 	"context"
+	"fmt"
 	"reflect"
 	"time"
 
 	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/csi/service/common"
 
 	cnsvolumemetadatav1alpha1 "sigs.k8s.io/vsphere-csi-driver/v3/pkg/apis/cnsoperator/cnsvolumemetadata/v1alpha1"
 	cnsconfig "sigs.k8s.io/vsphere-csi-driver/v3/pkg/common/config"
@@ -138,6 +141,14 @@ func PvcsiFullSync(ctx context.Context, metadataSyncer *metadataSyncInformer) er
 		}
 	}
 
+	// Set csi.vsphere.tanzu-kubernetes-cluster annotation on the Supervisor PVC which is
+	// requested from TKC Cluster
+	err = setTanzuClusterIDOnSupervisorPVC(ctx, metadataSyncer, supervisorNamespace)
+	if err != nil {
+		log.Errorf("FullSync: Failed to create CnsVolumeMetadataList from guest cluster. Err: %v", err)
+		return err
+	}
+
 	log.Infof("FullSync: End")
 	return nil
 }
@@ -233,6 +244,42 @@ func createCnsVolumeMetadataList(ctx context.Context, metadataSyncer *metadataSy
 	return nil
 }
 
+func setTanzuClusterIDOnSupervisorPVC(ctx context.Context, metadataSyncer *metadataSyncInformer,
+	supervisorNamespace string) error {
+	log := logger.GetLogger(ctx)
+	log.Debugf("FullSync: Querying guest cluster API server for all PV objects.")
+	pvList, err := getPVsInBoundAvailableOrReleased(ctx, metadataSyncer)
+	if err != nil {
+		log.Errorf("FullSync: Failed to get PVs from guest cluster. Err: %v", err)
+		return err
+	}
+	for _, pv := range pvList {
+		if pv.Spec.ClaimRef != nil && pv.Status.Phase == v1.VolumeBound {
+			svPVC, err := metadataSyncer.supervisorClient.CoreV1().PersistentVolumeClaims(supervisorNamespace).
+				Get(ctx, pv.Spec.CSI.VolumeHandle, metav1.GetOptions{})
+			if err != nil {
+				msg := fmt.Sprintf("failed to retrieve supervisor PVC %q in %q namespace. Error: %+v",
+					pv.Spec.CSI.VolumeHandle, supervisorNamespace, err)
+				log.Error(msg)
+				continue
+			}
+			if svPVC.Annotations[common.AnnTanzuGuestClusterOwner] != "" {
+
+				svPVC.Annotations[common.AnnTanzuGuestClusterOwner] = metadataSyncer.configInfo.Cfg.GC.TanzuKubernetesClusterUID
+				_, err = metadataSyncer.supervisorClient.CoreV1().PersistentVolumeClaims(supervisorNamespace).Update(
+					ctx, svPVC, metav1.UpdateOptions{})
+				if err != nil {
+					msg := fmt.Sprintf("failed to update supervisor PVC: %q with annoation %q in %q namespace. Error: %+v",
+						pv.Spec.CSI.VolumeHandle, common.AnnTanzuGuestClusterOwner, supervisorNamespace, err)
+					log.Error(msg)
+					continue
+				}
+			}
+		}
+	}
+	return nil
+}
+
 // compareCnsVolumeMetadatas compares input cnsvolumemetadata objects
 // and returns false if their labels are not deeply equal.
 func compareCnsVolumeMetadatas(guestObject *cnsvolumemetadatav1alpha1.CnsVolumeMetadataSpec,