Repository: kubernetes-csi/external-provisioner
Pull request #421 (2020-03-11 07:52:29)

subset of supported topology keys are not working for waitforfirstconsumer
**What type of PR is this?**
/kind feature


**What this PR does / why we need it:**

fixes:- https://github.com/kubernetes-csi/external-provisioner/issues/420

If waitforfirstconsumer is used and if the driver supports multiple topology  keys,
mentioning single key(or less than supported keys) in the storageclass is not working.
The csi-provisioner gets list of all the topology keys from the csinode object
and gets the values for those keys from the node object. Now this key-value term
is matched against the topology mentioned in the storageclass and it should match
entirely which is not correct as even if topology mentioned in the storageclass
is subset of the key-value term supported by the driver, then the node is a valid condidate.

Instead of strictly checking for the equality of the selected node's topology with
the allowedTopologies mentioned in the storageclass, we should check if allowedTopologies
is subset of selected node's topology.

**Does this PR introduce a user-facing change?:**
```release-note
We can now provide few keys in the allowedtopologies of all the keys supported by the driver in late binding case.
```

k8s-ci-robot: @pawanpraka1: Adding the "do-not-merge/release-note-label-needed" label because no release-note block was detected, please follow our [release note process](https://git.k8s.io/community/contributors/guide/release-notes.md) to remove it.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
k8s-ci-robot: Thanks for your pull request. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

:memo: **Please follow instructions at <https://git.k8s.io/community/CLA.md#the-contributor-license-agreement> to sign the CLA.**

It may take a couple minutes for the CLA signature to be fully registered; after that, please reply here with a new comment and we'll verify.  Thanks.

---

- If you've already signed a CLA, it's possible we don't have your GitHub username or you're using a different email address.  Check your existing CLA data and verify that your [email is set on your git commits](https://help.github.com/articles/setting-your-email-in-git/).
- If you signed the CLA as a corporation, please sign in with your organization's credentials at <https://identity.linuxfoundation.org/projects/cncf> to be authorized.
- If you have done the above and are still having issues with the CLA being reported as unsigned, please log a ticket with the Linux Foundation Helpdesk: <https://support.linuxfoundation.org/>
- Should you encounter any issues with the Linux Foundation Helpdesk, send a message to the backup e-mail support address at: login-issues@jira.linuxfoundation.org

<!-- need_sender_cla -->

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
	
k8s-ci-robot: Welcome @pawanpraka1! <br><br>It looks like this is your first PR to <a href='https://github.com/kubernetes-csi/external-provisioner'>kubernetes-csi/external-provisioner</a> 🎉. Please refer to our [pull request process documentation](https://git.k8s.io/community/contributors/guide/pull-requests.md) to help your PR have a smooth ride to approval. <br><br>You will be prompted by a bot to use commands during the review process. Do not be afraid to follow the prompts! It is okay to experiment. [Here is the bot commands documentation](https://go.k8s.io/bot-commands). <br><br>You can also check if kubernetes-csi/external-provisioner has [its own contribution guidelines](https://github.com/kubernetes-csi/external-provisioner/tree/master/CONTRIBUTING.md). <br><br>You may want to refer to our [testing guide](https://git.k8s.io/community/contributors/devel/sig-testing/testing.md) if you run into trouble with your tests not passing. <br><br>If you are having difficulty getting your pull request seen, please follow the [recommended escalation practices](https://github.com/kubernetes/community/blob/master/contributors/guide/pull-requests.md#why-is-my-pull-request-not-getting-reviewed). Also, for tips and tricks in the contribution process you may want to read the [Kubernetes contributor cheat sheet](https://git.k8s.io/community/contributors/guide/contributor-cheatsheet/README.md). We want to make sure your contribution gets all the attention it needs! <br><br>Thank you, and welcome to Kubernetes. :smiley:
k8s-ci-robot: Hi @pawanpraka1. Thanks for your PR.

I'm waiting for a [kubernetes-csi](https://github.com/orgs/kubernetes-csi/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://git.k8s.io/community/community-membership.md#member) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=kubernetes-csi%2Fexternal-provisioner).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>

k8s-ci-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-csi/external-provisioner/pull/421#" title="Author self-approved">pawanpraka1</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **jsafrane**
You can assign the PR to them by writing `/assign @jsafrane` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-csi%2Fexternal-provisioner).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/kubernetes-csi/external-provisioner/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["jsafrane"]} -->
pawanpraka1: I signed it
pawanpraka1: I signed it
pawanpraka1: I signed it
pawanpraka1: I signed it
msau42: /ok-to-test
/assign @verult 
pawanpraka1: /assign @msau42 
msau42: /assign @pohly 
k8s-ci-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-csi/external-provisioner/pull/421#" title="Author self-approved">pawanpraka1</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **msau42**
You can assign the PR to them by writing `/assign @msau42` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-csi%2Fexternal-provisioner).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/kubernetes-csi/external-provisioner/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["msau42"]} -->
verult: /lgtm

Just thinking out loud and throwing in another perspective: What happens on driver upgrade when the new version of the driver introduces a new key? Say there was an `AllowedTopologies` restricting only on the old key, and the admin forgets to update the StorageClass to include the new key. With the subset logic, k8s will match all previously matched nodes before the upgrade. IMO this is great and better than before, where the volume provision will fail.
msau42: Can you also add a release note for this?
k8s-ci-robot: @pawanpraka1: The label(s) `kind/release-note` cannot be applied, because the repository doesn't have them

<details>

In response to [this](https://github.com/kubernetes-csi/external-provisioner/pull/421):

> /kind bug
>fixes:- https://github.com/kubernetes-csi/external-provisioner/issues/420
>
>If waitforfirstconsumer is used and if the driver supports multiple topology  keys,
>mentioning single key(or less than supported keys) in the storageclass is not working.
>The csi-provisioner gets list of all the topology keys from the csinode object
>and gets the values for those keys from the node object. Now this key-value term
>is matched against the topology mentioned in the storageclass and it should match
>entirely which is not correct as even if topology mentioned in the storageclass
>is subset of the key-value term supported by the driver, then the node is a valid condidate.
>
>Instead of strictly checking for the equality of the selected node's topology with
>the allowedTopologies mentioned in the storageclass, we should check if allowedTopologies
>is subset of selected node's topology.
>
>
>/kind documentation
>/kind release-note
>When we are using few keys of all the supported keys by the driver in the storageclass allowedtopologies, the volume provisioning was failing. There was a bug in matching the allowedtopologies mentioned in the storageclass with the topologies supported by the driver. It was assuming all the keys supported by the driver to be present in the allowed topologies. Fixed this relaxing this assumption. 


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
k8s-ci-robot: @pawanpraka1: The label(s) `kind/documentatinon` cannot be applied, because the repository doesn't have them

<details>

In response to [this](https://github.com/kubernetes-csi/external-provisioner/pull/421):

>What type of PR is this?
> /kind bug
>/kind documentation
>
>What this PR does / why we need it:
>
>
>fixes:- https://github.com/kubernetes-csi/external-provisioner/issues/420
>
>If waitforfirstconsumer is used and if the driver supports multiple topology  keys,
>mentioning single key(or less than supported keys) in the storageclass is not working.
>The csi-provisioner gets list of all the topology keys from the csinode object
>and gets the values for those keys from the node object. Now this key-value term
>is matched against the topology mentioned in the storageclass and it should match
>entirely which is not correct as even if topology mentioned in the storageclass
>is subset of the key-value term supported by the driver, then the node is a valid condidate.
>
>Instead of strictly checking for the equality of the selected node's topology with
>the allowedTopologies mentioned in the storageclass, we should check if allowedTopologies
>is subset of selected node's topology.
>
>Special notes for your reviewer:
>
>Does this PR introduce a user-facing change?:
>/kind documentatinon
>When we are using few keys of all the supported keys by the driver in the storageclass allowedtopologies, the volume provisioning was failing. There was a bug in matching the allowedtopologies mentioned in the storageclass with the topologies supported by the driver. It was assuming all the keys supported by the driver to be present in the allowed topologies. Fixed this relaxing this assumption. 


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
k8s-ci-robot: @pawanpraka1: The label(s) `kind/documentatinon` cannot be applied, because the repository doesn't have them

<details>

In response to [this](https://github.com/kubernetes-csi/external-provisioner/pull/421):

>**What type of PR is this?**
> /kind bug
>/kind documentation
>
>**What this PR does / why we need it:**
>
>fixes:- https://github.com/kubernetes-csi/external-provisioner/issues/420
>
>If waitforfirstconsumer is used and if the driver supports multiple topology  keys,
>mentioning single key(or less than supported keys) in the storageclass is not working.
>The csi-provisioner gets list of all the topology keys from the csinode object
>and gets the values for those keys from the node object. Now this key-value term
>is matched against the topology mentioned in the storageclass and it should match
>entirely which is not correct as even if topology mentioned in the storageclass
>is subset of the key-value term supported by the driver, then the node is a valid condidate.
>
>Instead of strictly checking for the equality of the selected node's topology with
>the allowedTopologies mentioned in the storageclass, we should check if allowedTopologies
>is subset of selected node's topology.
>
>**Special notes for your reviewer:**
>
>**Does this PR introduce a user-facing change?:**
>/kind documentatinon
>When we are using few keys of all the supported keys by the driver in the storageclass allowedtopologies, the volume provisioning was failing. There was a bug in matching the allowedtopologies mentioned in the storageclass with the topologies supported by the driver. It was assuming all the keys supported by the driver to be present in the allowed topologies. Fixed this relaxing this assumption. 


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
pawanpraka1: > Can you also add a release note for this?

done @msau42 .
msau42: /kind feature
/remove-kind documentation
k8s-ci-robot: New changes are detected. LGTM label has been removed.
msau42: /lgtm
/approve
Thanks for working on this!
k8s-ci-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-csi/external-provisioner/pull/421#issuecomment-607337747" title="Approved">msau42</a>*, *<a href="https://github.com/kubernetes-csi/external-provisioner/pull/421#" title="Author self-approved">pawanpraka1</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-csi%2Fexternal-provisioner).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/kubernetes-csi/external-provisioner/blob/master/OWNERS)~~ [msau42]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
pawanpraka1: @msau42  travis has passed but status has not updated yet, Please check.
msau42: Can you try force pushing your branch again? I'm not sure why travis reporting is stuck
pawanpraka1: /retest
pawanpraka1: @msau42 can you restart the travis build?
msau42: I'm not able to restart it, I think it is triggered based on git pushes. Maybe try squashsing your commits and then force pushing?

Patch:
diff --git a/CHANGELOG-1.1.md b/CHANGELOG-1.1.md
index 944bf5988..1856aa111 100644
--- a/CHANGELOG-1.1.md
+++ b/CHANGELOG-1.1.md
@@ -15,7 +15,7 @@
 * [In-tree storage plugin to CSI Driver Migration](https://github.com/kubernetes/enhancements/blob/master/keps/sig-storage/20190129-csi-migration.md) is now alpha ([#253](https://github.com/kubernetes-csi/external-provisioner/pull/253))
 * The external provisioner now tries to connect to the CSI driver indefinitely ([#234](https://github.com/kubernetes-csi/external-provisioner/pull/234))
 * A new --timeout parameter has been added for CSI operations ([#230](https://github.com/kubernetes-csi/external-provisioner/pull/230))
-* README.md has been significantly enhanced ([#249](https://github.com/kubernetes-csi/external-provisioner/pull/249))
+* README.md has been signficantly enhanced ([#249](https://github.com/kubernetes-csi/external-provisioner/pull/249))
 * Add support for  Lease based leader election. Enable this by setting
   `--leader-election-type=leases` ([#261](https://github.com/kubernetes-csi/external-provisioner/pull/261))
 
diff --git a/README.md b/README.md
index 29b073caa..f04970a65 100644
--- a/README.md
+++ b/README.md
@@ -61,13 +61,11 @@ Note that the external-provisioner does not scale with more replicas. Only one e
 
 * `--worker-threads <num>`: Number of simultaneously running `ControllerCreateVolume` and `ControllerDeleteVolume` operations. Default value is `100`.
 
-* `--cloning-protection-threads <num>`: Number of simultaniously running threads, handling cloning finalizer removal. Defaults to `1`.
-
 * `--metrics-address`: The TCP network address where the prometheus metrics endpoint will run (example: `:8080` which corresponds to port 8080 on local host). The default is empty string, which means metrics endpoint is disabled.
 
 * `--metrics-path`: The HTTP path where prometheus metrics will be exposed. Default is `/metrics`.
 
-* `--extra-create-metadata`: Enables the injection of extra PVC and PV metadata as parameters when calling `CreateVolume` on the driver (keys: "csi.storage.k8s.io/pvc/name", "csi.storage.k8s.io/pvc/namespace", "csi.storage.k8s.io/pv/name")
+* `--extra-create-metadata`: Enables the injection of extra PVC and PV metadata as parameters when calling `CreateVolume` on the driver (keys: "csi.storage.k8s.io/pvc/name", "csi.storage.k8s.io/pvc/namespace", "csi.storage.k8s.io/pv/name") 
 
 #### Other recognized arguments
 * `--feature-gates <gates>`: A set of comma separated `<feature-name>=<true|false>` pairs that describe feature gates for alpha/experimental features. See [list of features](#feature-status) or `--help` output for list of recognized features. Example: `--feature-gates Topology=true` to enable Topology feature that's disabled by default.
diff --git a/cmd/csi-provisioner/csi-provisioner.go b/cmd/csi-provisioner/csi-provisioner.go
index 8bfdc4882..669a0a700 100644
--- a/cmd/csi-provisioner/csi-provisioner.go
+++ b/cmd/csi-provisioner/csi-provisioner.go
@@ -61,7 +61,6 @@ var (
 	retryIntervalStart   = flag.Duration("retry-interval-start", time.Second, "Initial retry interval of failed provisioning or deletion. It doubles with each failure, up to retry-interval-max.")
 	retryIntervalMax     = flag.Duration("retry-interval-max", 5*time.Minute, "Maximum retry interval of failed provisioning or deletion.")
 	workerThreads        = flag.Uint("worker-threads", 100, "Number of provisioner worker threads, in other words nr. of simultaneous CSI calls.")
-	finalizerThreads     = flag.Uint("cloning-protection-threads", 1, "Number of simultaniously running threads, handling cloning finalizer removal")
 	operationTimeout     = flag.Duration("timeout", 10*time.Second, "Timeout for waiting for creation or deletion of a volume")
 	_                    = deprecatedflags.Add("provisioner")
 
@@ -173,36 +172,13 @@ func main() {
 	timeStamp := time.Now().UnixNano() / int64(time.Millisecond)
 	identity := strconv.FormatInt(timeStamp, 10) + "-" + strconv.Itoa(rand.Intn(10000)) + "-" + provisionerName
 
-	factory := informers.NewSharedInformerFactory(clientset, ctrl.ResyncPeriodOfCsiNodeInformer)
-
-	// -------------------------------
-	// Listers
-	// Create informer to prevent hit the API server for all resource request
-	scLister := factory.Storage().V1().StorageClasses().Lister()
-	claimLister := factory.Core().V1().PersistentVolumeClaims().Lister()
-
-	var csiNodeLister storagelistersv1beta1.CSINodeLister
-	var nodeLister v1.NodeLister
-	if ctrl.SupportsTopology(pluginCapabilities) {
-		csiNodeLister = factory.Storage().V1beta1().CSINodes().Lister()
-		nodeLister = factory.Core().V1().Nodes().Lister()
-	}
-
-	// -------------------------------
-	// PersistentVolumeClaims informer
-	rateLimiter := workqueue.NewItemExponentialFailureRateLimiter(*retryIntervalStart, *retryIntervalMax)
-	claimQueue := workqueue.NewNamedRateLimitingQueue(rateLimiter, "claims")
-	claimInformer := factory.Core().V1().PersistentVolumeClaims().Informer()
-
-	// Setup options
 	provisionerOptions := []func(*controller.ProvisionController) error{
 		controller.LeaderElection(false), // Always disable leader election in provisioner lib. Leader election should be done here in the CSI provisioner level instead.
 		controller.FailedProvisionThreshold(0),
 		controller.FailedDeleteThreshold(0),
-		controller.RateLimiter(rateLimiter),
+		controller.RateLimiter(workqueue.NewItemExponentialFailureRateLimiter(*retryIntervalStart, *retryIntervalMax)),
 		controller.Threadiness(int(*workerThreads)),
 		controller.CreateProvisionedPVLimiter(workqueue.DefaultControllerRateLimiter()),
-		controller.ClaimsInformer(claimInformer),
 	}
 
 	translator := csitrans.New()
@@ -217,6 +193,17 @@ func main() {
 		provisionerOptions = append(provisionerOptions, controller.AdditionalProvisionerNames([]string{supportsMigrationFromInTreePluginName}))
 	}
 
+	// Create informer to prevent hit the API server for all resource request
+	factory := informers.NewSharedInformerFactory(clientset, ctrl.ResyncPeriodOfCsiNodeInformer)
+	scLister := factory.Storage().V1().StorageClasses().Lister()
+
+	var csiNodeLister storagelistersv1beta1.CSINodeLister
+	var nodeLister v1.NodeLister
+	if ctrl.SupportsTopology(pluginCapabilities) {
+		csiNodeLister = factory.Storage().V1beta1().CSINodes().Lister()
+		nodeLister = factory.Core().V1().Nodes().Lister()
+	}
+
 	// Create the provisioner: it implements the Provisioner interface expected by
 	// the controller
 	csiProvisioner := ctrl.NewCSIProvisioner(
@@ -236,7 +223,6 @@ func main() {
 		scLister,
 		csiNodeLister,
 		nodeLister,
-		claimLister,
 		*extraCreateMetadata,
 	)
 
@@ -248,13 +234,6 @@ func main() {
 		provisionerOptions...,
 	)
 
-	csiClaimController := ctrl.NewCloningProtectionController(
-		clientset,
-		claimLister,
-		claimInformer,
-		claimQueue,
-	)
-
 	run := func(context.Context) {
 		stopCh := context.Background().Done()
 		factory.Start(stopCh)
@@ -265,7 +244,6 @@ func main() {
 			}
 		}
 
-		go csiClaimController.Run(int(*finalizerThreads), stopCh)
 		provisionController.Run(wait.NeverStop)
 	}
 
diff --git a/pkg/controller/clone_controller.go b/pkg/controller/clone_controller.go
deleted file mode 100644
index f947495f9..000000000
--- a/pkg/controller/clone_controller.go
+++ /dev/null
@@ -1,206 +0,0 @@
-package controller
-
-import (
-	"fmt"
-	"time"
-
-	v1 "k8s.io/api/core/v1"
-	apierrs "k8s.io/apimachinery/pkg/api/errors"
-	"k8s.io/apimachinery/pkg/labels"
-	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
-	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/client-go/kubernetes"
-	corelisters "k8s.io/client-go/listers/core/v1"
-	"k8s.io/client-go/tools/cache"
-	"k8s.io/client-go/util/workqueue"
-	"k8s.io/klog"
-	"sigs.k8s.io/sig-storage-lib-external-provisioner/controller"
-)
-
-//
-// This package introduce a way to handle finalizers, related to in-progress PVC cloning. This is a two step approach:
-//
-// 1) PVC referenced as a data source is now updated with a finalizer `provisioner.storage.kubernetes.io/cloning-protection` during a ProvisionExt method call.
-// The detection of cloning in-progress is based on the assumption that a PVC with `spec.DataSource` pointing on a another PVC will go into `Pending` state.
-// The downside of this, is that fact that any other reason causing PVC to stay in the `Pending` state also blocks resource from deletion it from deletion
-//
-// 2) When cloning is finished for each PVC referencing the one as a data source,
-// this PVC will go from `Pending` to `Bound` state. That allows remove the finalizer.
-//
-
-// CloningProtectionController is storing all related interfaces
-// to handle cloning protection finalizer removal after CSI cloning is finished
-type CloningProtectionController struct {
-	client        kubernetes.Interface
-	claimLister   corelisters.PersistentVolumeClaimLister
-	claimInformer cache.SharedInformer
-	claimQueue    workqueue.RateLimitingInterface
-}
-
-// NewCloningProtectionController creates new controller for additional CSI claim protection capabilities
-func NewCloningProtectionController(
-	client kubernetes.Interface,
-	claimLister corelisters.PersistentVolumeClaimLister,
-	claimInformer cache.SharedInformer,
-	claimQueue workqueue.RateLimitingInterface,
-) *CloningProtectionController {
-	controller := &CloningProtectionController{
-		client:        client,
-		claimLister:   claimLister,
-		claimInformer: claimInformer,
-		claimQueue:    claimQueue,
-	}
-	return controller
-}
-
-// Run is a main CloningProtectionController handler
-func (p *CloningProtectionController) Run(threadiness int, stopCh <-chan struct{}) {
-	klog.Info("Starting CloningProtection controller")
-	defer utilruntime.HandleCrash()
-	defer p.claimQueue.ShutDown()
-
-	claimHandler := cache.ResourceEventHandlerFuncs{
-		AddFunc:    p.enqueueClaimUpadate,
-		UpdateFunc: func(_ interface{}, newObj interface{}) { p.enqueueClaimUpadate(newObj) },
-	}
-	p.claimInformer.AddEventHandlerWithResyncPeriod(claimHandler, controller.DefaultResyncPeriod)
-
-	for i := 0; i < threadiness; i++ {
-		go wait.Until(p.runClaimWorker, time.Second, stopCh)
-	}
-
-	go p.claimInformer.Run(stopCh)
-
-	klog.Infof("Started CloningProtection controller")
-	<-stopCh
-	klog.Info("Shutting down CloningProtection controller")
-}
-
-func (p *CloningProtectionController) runClaimWorker() {
-	for p.processNextClaimWorkItem() {
-	}
-}
-
-// processNextClaimWorkItem processes items from claimQueue
-func (p *CloningProtectionController) processNextClaimWorkItem() bool {
-	obj, shutdown := p.claimQueue.Get()
-	if shutdown {
-		return false
-	}
-
-	err := func(obj interface{}) error {
-		defer p.claimQueue.Done(obj)
-		var key string
-		var ok bool
-		if key, ok = obj.(string); !ok {
-			p.claimQueue.Forget(obj)
-			return fmt.Errorf("expected string in workqueue but got %#v", obj)
-		}
-
-		if err := p.syncClaimHandler(key); err != nil {
-			klog.Warningf("Retrying syncing claim %q after %v failures", key, p.claimQueue.NumRequeues(obj))
-			p.claimQueue.AddRateLimited(obj)
-		} else {
-			p.claimQueue.Forget(obj)
-		}
-
-		return nil
-	}(obj)
-
-	if err != nil {
-		utilruntime.HandleError(err)
-		return true
-	}
-
-	return true
-}
-
-// enqueueClaimUpadate takes a PVC obj and stores it into the claim work queue.
-func (p *CloningProtectionController) enqueueClaimUpadate(obj interface{}) {
-	new, ok := obj.(*v1.PersistentVolumeClaim)
-	if !ok {
-		utilruntime.HandleError(fmt.Errorf("expected claim but got %+v", new))
-		return
-	}
-
-	// Timestamp didn't appear
-	if new.DeletionTimestamp == nil {
-		return
-	}
-
-	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
-	if err != nil {
-		utilruntime.HandleError(err)
-		return
-	}
-
-	p.claimQueue.Add(key)
-}
-
-// syncClaimHandler gets the claim from informer's cache then calls syncClaim
-func (p *CloningProtectionController) syncClaimHandler(key string) error {
-	namespace, name, err := cache.SplitMetaNamespaceKey(key)
-	if err != nil {
-		utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
-		return nil
-	}
-
-	claim, err := p.claimLister.PersistentVolumeClaims(namespace).Get(name)
-	if err != nil {
-		if apierrs.IsNotFound(err) {
-			utilruntime.HandleError(fmt.Errorf("Item '%s' in work queue no longer exists", key))
-			return nil
-		}
-
-		return err
-	}
-
-	return p.syncClaim(claim)
-}
-
-// syncClaim removes finalizers from a PVC, when cloning is finished
-func (p *CloningProtectionController) syncClaim(claim *v1.PersistentVolumeClaim) error {
-	if !checkFinalizer(claim, pvcCloneFinalizer) {
-		return nil
-	}
-
-	// Checking for PVCs in the same namespace to have other states aside from Pending, which means that cloning is still in progress
-	pvcList, err := p.claimLister.PersistentVolumeClaims(claim.Namespace).List(labels.Everything())
-	if err != nil {
-		return err
-	}
-
-	// Check for pvc state with DataSource pointing to claim
-	for _, pvc := range pvcList {
-		if pvc.Spec.DataSource == nil {
-			continue
-		}
-
-		// Requeue when at least one PVC is still works on cloning
-		if pvc.Spec.DataSource.Kind == pvcKind &&
-			pvc.Spec.DataSource.Name == claim.Name &&
-			pvc.Status.Phase == v1.ClaimPending {
-			return fmt.Errorf("PVC '%s' is in 'Pending' state, cloning in progress", pvc.Name)
-		}
-	}
-
-	// Remove clone finalizer
-	finalizers := make([]string, 0)
-	for _, finalizer := range claim.ObjectMeta.Finalizers {
-		if finalizer != pvcCloneFinalizer {
-			finalizers = append(finalizers, finalizer)
-		}
-	}
-	claim.ObjectMeta.Finalizers = finalizers
-
-	if _, err = p.client.CoreV1().PersistentVolumeClaims(claim.Namespace).Update(claim); err != nil {
-		if !apierrs.IsNotFound(err) {
-			// Couldn't remove finalizer and the object still exists, the controller may
-			// try to remove the finalizer again on the next update
-			klog.Infof("failed to remove clone finalizer from PVC %v", claim.Name)
-			return err
-		}
-	}
-
-	return nil
-}
diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
index 2d8d1c15d..c6d038506 100644
--- a/pkg/controller/controller.go
+++ b/pkg/controller/controller.go
@@ -134,8 +134,6 @@ const (
 	annStorageProvisioner = "volume.beta.kubernetes.io/storage-provisioner"
 
 	snapshotNotBound = "snapshot %s not bound"
-
-	pvcCloneFinalizer = "provisioner.storage.kubernetes.io/cloning-protection"
 )
 
 var (
@@ -222,7 +220,6 @@ type csiProvisioner struct {
 	scLister                              storagelistersv1.StorageClassLister
 	csiNodeLister                         storagelistersv1beta1.CSINodeLister
 	nodeLister                            corelisters.NodeLister
-	claimLister                           corelisters.PersistentVolumeClaimLister
 	extraCreateMetadata                   bool
 }
 
@@ -287,7 +284,6 @@ func NewCSIProvisioner(client kubernetes.Interface,
 	scLister storagelistersv1.StorageClassLister,
 	csiNodeLister storagelistersv1beta1.CSINodeLister,
 	nodeLister corelisters.NodeLister,
-	claimLister corelisters.PersistentVolumeClaimLister,
 	extraCreateMetadata bool,
 ) controller.Provisioner {
 
@@ -310,7 +306,6 @@ func NewCSIProvisioner(client kubernetes.Interface,
 		scLister:                              scLister,
 		csiNodeLister:                         csiNodeLister,
 		nodeLister:                            nodeLister,
-		claimLister:                           claimLister,
 		extraCreateMetadata:                   extraCreateMetadata,
 	}
 	return provisioner
@@ -536,13 +531,6 @@ func (p *csiProvisioner) ProvisionExt(options controller.ProvisionOptions) (*v1.
 		req.VolumeContentSource = volumeContentSource
 	}
 
-	if options.PVC.Spec.DataSource != nil && rc.clone {
-		err = p.setCloneFinalizer(options.PVC)
-		if err != nil {
-			return nil, controller.ProvisioningNoChange, err
-		}
-	}
-
 	if p.supportsTopology() {
 		requirements, err := GenerateAccessibilityRequirements(
 			p.client,
@@ -723,21 +711,6 @@ func (p *csiProvisioner) ProvisionExt(options controller.ProvisionOptions) (*v1.
 	return pv, controller.ProvisioningFinished, nil
 }
 
-func (p *csiProvisioner) setCloneFinalizer(pvc *v1.PersistentVolumeClaim) error {
-	claim, err := p.claimLister.PersistentVolumeClaims(pvc.Namespace).Get(pvc.Spec.DataSource.Name)
-	if err != nil {
-		return err
-	}
-
-	if !checkFinalizer(claim, pvcCloneFinalizer) {
-		claim.Finalizers = append(claim.Finalizers, pvcCloneFinalizer)
-		_, err := p.client.CoreV1().PersistentVolumeClaims(claim.Namespace).Update(claim)
-		return err
-	}
-
-	return nil
-}
-
 func (p *csiProvisioner) supportsTopology() bool {
 	return SupportsTopology(p.pluginCapabilities)
 }
@@ -792,7 +765,7 @@ func (p *csiProvisioner) getVolumeContentSource(options controller.ProvisionOpti
 // getPVCSource verifies DataSource.Kind of type PersistentVolumeClaim, making sure that the requested PVC is available/ready
 // returns the VolumeContentSource for the requested PVC
 func (p *csiProvisioner) getPVCSource(options controller.ProvisionOptions) (*csi.VolumeContentSource, error) {
-	sourcePVC, err := p.claimLister.PersistentVolumeClaims(options.PVC.Namespace).Get(options.PVC.Spec.DataSource.Name)
+	sourcePVC, err := p.client.CoreV1().PersistentVolumeClaims(options.PVC.Namespace).Get(options.PVC.Spec.DataSource.Name, metav1.GetOptions{})
 	if err != nil {
 		return nil, fmt.Errorf("error getting PVC %s (namespace %q) from api server: %v", options.PVC.Spec.DataSource.Name, options.PVC.Namespace, err)
 	}
@@ -1272,7 +1245,7 @@ func isFinalError(err error) bool {
 }
 
 func cleanupVolume(p *csiProvisioner, delReq *csi.DeleteVolumeRequest, provisionerCredentials map[string]string) error {
-	var err error
+	var err error = nil
 	delReq.Secrets = provisionerCredentials
 	ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
 	defer cancel()
@@ -1284,12 +1257,3 @@ func cleanupVolume(p *csiProvisioner, delReq *csi.DeleteVolumeRequest, provision
 	}
 	return err
 }
-
-func checkFinalizer(obj metav1.Object, finalizer string) bool {
-	for _, f := range obj.GetFinalizers() {
-		if f == finalizer {
-			return true
-		}
-	}
-	return false
-}
diff --git a/pkg/controller/topology.go b/pkg/controller/topology.go
index 878789d28..487a1d7f4 100644
--- a/pkg/controller/topology.go
+++ b/pkg/controller/topology.go
@@ -200,7 +200,7 @@ func GenerateAccessibilityRequirements(
 				allowedTopologiesFlatten := flatten(allowedTopologies)
 				found := false
 				for _, t := range allowedTopologiesFlatten {
-					if t.equal(selectedTopology) {
+					if t.subset(selectedTopology) {
 						found = true
 						break
 					}
@@ -480,7 +480,7 @@ func sortAndShift(terms []topologyTerm, primary topologyTerm, shiftIndex uint32)
 		preferredTerms = append(terms[shiftIndex:], terms[:shiftIndex]...)
 	} else {
 		for i, t := range terms {
-			if t.equal(primary) {
+			if t.subset(primary) {
 				preferredTerms = append(terms[i:], terms[:i]...)
 				break
 			}
@@ -562,6 +562,17 @@ func (t topologyTerm) less(other topologyTerm) bool {
 	return t.hash() < other.hash()
 }
 
+func (t topologyTerm) subset(other topologyTerm) bool {
+	for key, tv := range t {
+		v, ok := other[key]
+		if !ok || v != tv {
+			return false
+		}
+	}
+
+	return true
+}
+
 func (t topologyTerm) equal(other topologyTerm) bool {
 	return t.hash() == other.hash()
 }


Test patch:
diff --git a/pkg/controller/clone_controller_test.go b/pkg/controller/clone_controller_test.go
deleted file mode 100644
index aa20b3e4c..000000000
--- a/pkg/controller/clone_controller_test.go
+++ /dev/null
@@ -1,199 +0,0 @@
-package controller
-
-import (
-	"context"
-	"fmt"
-	"testing"
-	"time"
-
-	v1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
-	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
-	"k8s.io/client-go/informers"
-	fakeclientset "k8s.io/client-go/kubernetes/fake"
-	"k8s.io/client-go/util/workqueue"
-)
-
-var requestedBytes int64 = 1000
-var fakeSc1 string = "fake-sc-1"
-
-const (
-	srcName      = "clone-source-pvc"
-	dstName      = "destination-pvc"
-	srcNamespace = "fake-pvc-namespace"
-	pvName       = "test-testi"
-)
-
-func pvcFinalizers(pvc *v1.PersistentVolumeClaim, finalizers ...string) *v1.PersistentVolumeClaim {
-	pvc.Finalizers = append(pvc.Finalizers, finalizers...)
-	return pvc
-}
-
-func pvcDataSourceClone(srcName string, pvc *v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
-	apiGr := ""
-	pvc.Spec.DataSource = &v1.TypedLocalObjectReference{
-		APIGroup: &apiGr,
-		Kind:     pvcKind,
-		Name:     srcName,
-	}
-	return pvc
-}
-
-func pvcNamed(name string, pvc *v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
-	pvc.Name = name
-	return pvc
-}
-
-func pvcNamespaced(namespace string, pvc *v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
-	pvc.Namespace = namespace
-	return pvc
-}
-
-func pvcPhase(phase v1.PersistentVolumeClaimPhase, pvc *v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
-	pvc.Status.Phase = phase
-	return pvc
-}
-
-func baseClaim() *v1.PersistentVolumeClaim {
-	return fakeClaim(srcName, srcNamespace, "fake-claim-uid", requestedBytes, pvName, v1.ClaimBound, &fakeSc1, "").DeepCopy()
-}
-
-func pvcDeletionMarked(pvc *v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
-	timeX := metav1.NewTime(time.Now())
-	pvc.DeletionTimestamp = &timeX
-	return pvc
-}
-
-// TestCloneFinalizerRemoval tests create volume clone
-func TestCloneFinalizerRemoval(t *testing.T) {
-	testcases := map[string]struct {
-		initialClaims   []runtime.Object
-		cloneSource     runtime.Object
-		expectFinalizer bool
-		expectError     error
-	}{
-		"delete source pvc with no cloning in progress": {
-			cloneSource:   pvcFinalizers(baseClaim(), pvcCloneFinalizer),
-			initialClaims: []runtime.Object{pvcDataSourceClone(srcName, pvcNamed(dstName, baseClaim()))},
-		},
-		"delete source pvc when destination pvc status is claim pending": {
-			cloneSource:     pvcFinalizers(baseClaim(), pvcCloneFinalizer),
-			initialClaims:   []runtime.Object{pvcPhase(v1.ClaimPending, pvcDataSourceClone(srcName, pvcNamed(dstName, baseClaim())))},
-			expectFinalizer: true,
-			expectError:     fmt.Errorf("PVC '%s' is in 'Pending' state, cloning in progress", dstName),
-		},
-		"delete source pvc when at least one destination pvc status is claim pending": {
-			cloneSource: pvcFinalizers(baseClaim(), pvcCloneFinalizer),
-			initialClaims: []runtime.Object{
-				pvcDataSourceClone(srcName, pvcNamed(dstName, baseClaim())),
-				pvcPhase(v1.ClaimPending, pvcDataSourceClone(srcName, pvcNamed(dstName+"1", baseClaim()))),
-			},
-			expectFinalizer: true,
-			expectError:     fmt.Errorf("PVC '%s' is in 'Pending' state, cloning in progress", dstName+"1"),
-		},
-		"delete source pvc located in another namespace should not block": {
-			cloneSource:   pvcNamespaced(srcNamespace+"1", pvcFinalizers(baseClaim(), pvcCloneFinalizer)),
-			initialClaims: []runtime.Object{pvcPhase(v1.ClaimPending, pvcDataSourceClone(srcName, pvcNamed(dstName+"1", baseClaim())))},
-		},
-		"delete source pvc which is not cloned by any other pvc": {
-			cloneSource: pvcFinalizers(baseClaim(), pvcCloneFinalizer),
-		},
-		"delete source pvc without finalizer": {
-			cloneSource: baseClaim(),
-		},
-	}
-
-	for k, tc := range testcases {
-		tc := tc
-		t.Run(k, func(t *testing.T) {
-			t.Parallel()
-
-			objects := append(tc.initialClaims, tc.cloneSource)
-			clientSet := fakeclientset.NewSimpleClientset(objects...)
-			cloningProtector := fakeCloningProtector(clientSet, objects...)
-
-			// Simulate Delete behavior
-			claim := pvcDeletionMarked(tc.cloneSource.(*v1.PersistentVolumeClaim))
-			err := cloningProtector.syncClaim(claim)
-
-			// Get updated claim after reconcile finish
-			claim, _ = clientSet.CoreV1().PersistentVolumeClaims(claim.Namespace).Get(claim.Name, metav1.GetOptions{})
-
-			// Check finalizers removal
-			if tc.expectFinalizer && !checkFinalizer(claim, pvcCloneFinalizer) {
-				t.Errorf("Claim finalizer was expected to be found on: %s", claim.Name)
-			} else if !tc.expectFinalizer && checkFinalizer(claim, pvcCloneFinalizer) {
-				t.Errorf("Claim finalizer was not expected to be found on: %s", claim.Name)
-			}
-
-			if tc.expectError == nil && err != nil {
-				t.Errorf("Caught an unexpected error during 'syncClaim' run: %s", err)
-			} else if tc.expectError != nil && err == nil {
-				t.Errorf("Expected error during 'syncClaim' run, got nil: %s", tc.expectError)
-			} else if tc.expectError != nil && err != nil && tc.expectError.Error() != err.Error() {
-				t.Errorf("Unexpected error during 'syncClaim' run:\n\t%s\nExpected:\n\t%s", err, tc.expectError)
-			}
-		})
-	}
-
-}
-
-// TestEnqueueClaimUpadate ensure that PVCs will be processed for finalizer removal only on deletionTimestamp being set on the resource
-func TestEnqueueClaimUpadate(t *testing.T) {
-	testcases := map[string]struct {
-		claim    *v1.PersistentVolumeClaim
-		queueLen int
-	}{
-		"enqueue claim with deletionTimestamp": {
-			claim:    pvcDeletionMarked(baseClaim()),
-			queueLen: 1,
-		},
-		"enqueue claim without deletionTimestamp": {
-			claim:    baseClaim(),
-			queueLen: 0,
-		},
-	}
-
-	for k, tc := range testcases {
-		tc := tc
-		t.Run(k, func(t *testing.T) {
-			t.Parallel()
-
-			objects := []runtime.Object{}
-			clientSet := fakeclientset.NewSimpleClientset(objects...)
-			cloningProtector := fakeCloningProtector(clientSet, objects...)
-
-			// Simulate queue behavior
-			cloningProtector.enqueueClaimUpadate(tc.claim)
-
-			if cloningProtector.claimQueue.Len() != tc.queueLen {
-				t.Errorf("claimQueue should contain %d items, got: %d", tc.queueLen, cloningProtector.claimQueue.Len())
-			}
-		})
-	}
-}
-
-func fakeCloningProtector(client *fakeclientset.Clientset, objects ...runtime.Object) *CloningProtectionController {
-	utilruntime.ReallyCrash = false
-
-	informerFactory := informers.NewSharedInformerFactory(client, 1*time.Second)
-	claimInformer := informerFactory.Core().V1().PersistentVolumeClaims().Informer()
-	claimLister := informerFactory.Core().V1().PersistentVolumeClaims().Lister()
-	rateLimiter := workqueue.NewItemExponentialFailureRateLimiter(time.Second, 2*time.Second)
-	claimQueue := workqueue.NewNamedRateLimitingQueue(rateLimiter, "claims")
-
-	for _, claim := range objects {
-		claimInformer.GetStore().Add(claim)
-	}
-
-	informerFactory.WaitForCacheSync(context.TODO().Done())
-	go informerFactory.Start(context.TODO().Done())
-
-	return NewCloningProtectionController(
-		client,
-		claimLister,
-		claimInformer,
-		claimQueue,
-	)
-}
diff --git a/pkg/controller/controller_test.go b/pkg/controller/controller_test.go
index 6fcd9edc2..f045b4a73 100644
--- a/pkg/controller/controller_test.go
+++ b/pkg/controller/controller_test.go
@@ -404,7 +404,7 @@ func TestCreateDriverReturnsInvalidCapacityDuringProvision(t *testing.T) {
 
 	pluginCaps, controllerCaps := provisionCapabilities()
 	csiProvisioner := NewCSIProvisioner(nil, 5*time.Second, "test-provisioner", "test",
-		5, csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil, nil, nil, nil, false)
+		5, csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil, nil, nil, false)
 
 	// Requested PVC with requestedBytes storage
 	deletePolicy := v1.PersistentVolumeReclaimDelete
@@ -1669,7 +1669,7 @@ func runProvisionTest(t *testing.T, k string, tc provisioningTestcase, requested
 
 	pluginCaps, controllerCaps := provisionCapabilities()
 	csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5, csiConn.conn,
-		nil, provisionDriverName, pluginCaps, controllerCaps, supportsMigrationFromInTreePluginName, false, csitrans.New(), nil, nil, nil, nil, tc.withExtraMetadata)
+		nil, provisionDriverName, pluginCaps, controllerCaps, supportsMigrationFromInTreePluginName, false, csitrans.New(), nil, nil, nil, tc.withExtraMetadata)
 
 	out := &csi.CreateVolumeResponse{
 		Volume: &csi.Volume{
@@ -2405,7 +2405,7 @@ func TestProvisionFromSnapshot(t *testing.T) {
 
 		pluginCaps, controllerCaps := provisionFromSnapshotCapabilities()
 		csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5, csiConn.conn,
-			client, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil, nil, nil, nil, false)
+			client, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil, nil, nil, false)
 
 		out := &csi.CreateVolumeResponse{
 			Volume: &csi.Volume{
@@ -2575,11 +2575,11 @@ func TestProvisionWithTopologyEnabled(t *testing.T) {
 
 			clientSet := fakeclientset.NewSimpleClientset(nodes, csiNodes)
 
-			scLister, csiNodeLister, nodeLister, claimLister, stopChan := listers(clientSet)
+			scLister, csiNodeLister, nodeLister, stopChan := listers(clientSet)
 			defer close(stopChan)
 
 			csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5,
-				csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), scLister, csiNodeLister, nodeLister, claimLister, false)
+				csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), scLister, csiNodeLister, nodeLister, false)
 
 			pv, err := csiProvisioner.Provision(controller.ProvisionOptions{
 				StorageClass: &storagev1.StorageClass{},
@@ -2634,7 +2634,7 @@ func TestProvisionWithTopologyDisabled(t *testing.T) {
 	clientSet := fakeclientset.NewSimpleClientset()
 	pluginCaps, controllerCaps := provisionWithTopologyCapabilities()
 	csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5,
-		csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil, nil, nil, nil, false)
+		csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil, nil, nil, false)
 
 	out := &csi.CreateVolumeResponse{
 		Volume: &csi.Volume{
@@ -2804,9 +2804,9 @@ func runDeleteTest(t *testing.T, k string, tc deleteTestcase) {
 	}
 
 	pluginCaps, controllerCaps := provisionCapabilities()
-	scLister, _, _, _, _ := listers(clientSet)
+	scLister, _, _, _ := listers(clientSet)
 	csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5,
-		csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), scLister, nil, nil, nil, false)
+		csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), scLister, nil, nil, false)
 
 	err = csiProvisioner.Delete(tc.persistentVolume)
 	if tc.expectErr && err == nil {
@@ -2901,14 +2901,12 @@ func TestProvisionFromPVC(t *testing.T) {
 		restoredVolSizeBig   bool                     // set to request a smaller volSize than source PVC, default false
 		expectedPVSpec       *pvSpec                  // set to expected PVSpec on success, for deep comparison, default nil
 		cloneUnsupported     bool                     // set to state clone feature not supported in capabilities, default false
-		expectFinalizers     bool                     // while set, expects clone protection finalizers to be set on a PVC
 		sourcePVStatusPhase  v1.PersistentVolumePhase // set to change source PV Status.Phase, default "Bound"
 		expectErr            bool                     // set to state, test is expected to return errors, default false
 	}{
 		"provision with pvc data source": {
-			clonePVName:      pvName,
-			volOpts:          generatePVCForProvisionFromPVC(srcNamespace, srcName, fakeSc1, requestedBytes, ""),
-			expectFinalizers: true,
+			clonePVName: pvName,
+			volOpts:     generatePVCForProvisionFromPVC(srcNamespace, srcName, fakeSc1, requestedBytes, ""),
 			expectedPVSpec: &pvSpec{
 				Name:          pvName,
 				ReclaimPolicy: v1.PersistentVolumeReclaimDelete,
@@ -2940,7 +2938,6 @@ func TestProvisionFromPVC(t *testing.T) {
 		"provision with pvc data source destination too small": {
 			clonePVName:          pvName,
 			volOpts:              generatePVCForProvisionFromPVC(srcNamespace, srcName, fakeSc1, requestedBytes+1, ""),
-			expectFinalizers:     true,
 			restoredVolSizeSmall: true,
 			expectErr:            true,
 		},
@@ -3030,9 +3027,9 @@ func TestProvisionFromPVC(t *testing.T) {
 			expectErr:   true,
 		},
 		"provision block data source is block": {
-			clonePVName:      blockModePVName,
-			volOpts:          generatePVCForProvisionFromPVC(srcNamespace, blockModePVName, fakeSc1, requestedBytes, "block"),
-			expectFinalizers: true,
+			clonePVName: blockModePVName,
+			volOpts:     generatePVCForProvisionFromPVC(srcNamespace, blockModePVName, fakeSc1, requestedBytes, "block"),
+			expectErr:   false,
 		},
 		"provision block but data source is filesystem": {
 			clonePVName: filesystemPVName,
@@ -3045,16 +3042,14 @@ func TestProvisionFromPVC(t *testing.T) {
 			expectErr:   true,
 		},
 		"provision nil mode data source is nil": {
-			clonePVName:      pvName,
-			volOpts:          generatePVCForProvisionFromPVC(srcNamespace, srcName, fakeSc1, requestedBytes, ""),
-			expectFinalizers: true,
-			expectErr:        false,
+			clonePVName: pvName,
+			volOpts:     generatePVCForProvisionFromPVC(srcNamespace, srcName, fakeSc1, requestedBytes, ""),
+			expectErr:   false,
 		},
 		"provision filesystem data source is filesystem": {
-			clonePVName:      filesystemPVName,
-			volOpts:          generatePVCForProvisionFromPVC(srcNamespace, filesystemPVName, fakeSc1, requestedBytes, "filesystem"),
-			expectFinalizers: true,
-			expectErr:        false,
+			clonePVName: filesystemPVName,
+			volOpts:     generatePVCForProvisionFromPVC(srcNamespace, filesystemPVName, fakeSc1, requestedBytes, "filesystem"),
+			expectErr:   false,
 		},
 		"provision filesystem but data source is block": {
 			clonePVName: blockModePVName,
@@ -3062,10 +3057,9 @@ func TestProvisionFromPVC(t *testing.T) {
 			expectErr:   true,
 		},
 		"provision filesystem data source is nil": {
-			clonePVName:      pvName,
-			volOpts:          generatePVCForProvisionFromPVC(srcNamespace, srcName, fakeSc1, requestedBytes, "filesystem"),
-			expectFinalizers: true,
-			expectErr:        false,
+			clonePVName: pvName,
+			volOpts:     generatePVCForProvisionFromPVC(srcNamespace, srcName, fakeSc1, requestedBytes, "filesystem"),
+			expectErr:   false,
 		},
 	}
 
@@ -3223,29 +3217,15 @@ func TestProvisionFromPVC(t *testing.T) {
 				}).Return(&csi.DeleteVolumeResponse{}, nil).Times(1)
 			}
 
-			_, _, _, claimLister, _ := listers(clientSet)
-
 			// Phase: execute the test
 			csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5, csiConn.conn,
-				nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil, nil, nil, claimLister, false)
+				nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil, nil, nil, false)
 
 			pv, err = csiProvisioner.Provision(tc.volOpts)
 			if tc.expectErr && err == nil {
 				t.Errorf("test %q: Expected error, got none", k)
 			}
 
-			if tc.volOpts.PVC.Spec.DataSource != nil {
-				claim, _ := claimLister.PersistentVolumeClaims(tc.volOpts.PVC.Namespace).Get(tc.volOpts.PVC.Spec.DataSource.Name)
-				if claim != nil {
-					set := checkFinalizer(claim, pvcCloneFinalizer)
-					if tc.expectFinalizers && !set {
-						t.Errorf("Claim %s does not have clone protection finalizer set", claim.Name)
-					} else if !tc.expectFinalizers && set {
-						t.Errorf("Claim %s should not have clone protection finalizer set", claim.Name)
-					}
-				}
-			}
-
 			// Phase: process test responses
 			// process responses
 			if !tc.expectErr && err != nil {
@@ -3345,7 +3325,7 @@ func TestProvisionWithMigration(t *testing.T) {
 			pluginCaps, controllerCaps := provisionCapabilities()
 			csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner",
 				"test", 5, csiConn.conn, nil, driverName, pluginCaps, controllerCaps,
-				inTreePluginName, false, mockTranslator, nil, nil, nil, nil, false)
+				inTreePluginName, false, mockTranslator, nil, nil, nil, false)
 
 			// Set up return values (AnyTimes to avoid overfitting on implementation)
 
@@ -3505,7 +3485,7 @@ func TestDeleteMigration(t *testing.T) {
 			pluginCaps, controllerCaps := provisionCapabilities()
 			csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner",
 				"test", 5, csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "",
-				false, mockTranslator, nil, nil, nil, nil, false)
+				false, mockTranslator, nil, nil, nil, false)
 
 			// Set mock return values (AnyTimes to avoid overfitting on implementation details)
 			mockTranslator.EXPECT().IsPVMigratable(gomock.Any()).Return(tc.expectTranslation).AnyTimes()
diff --git a/pkg/controller/topology_test.go b/pkg/controller/topology_test.go
index dee0476cb..2479483ba 100644
--- a/pkg/controller/topology_test.go
+++ b/pkg/controller/topology_test.go
@@ -395,7 +395,7 @@ func TestStatefulSetSpreading(t *testing.T) {
 
 	kubeClient := fakeclientset.NewSimpleClientset(nodes, csiNodes)
 
-	_, csiNodeLister, nodeLister, _, stopChan := listers(kubeClient)
+	_, csiNodeLister, nodeLister, stopChan := listers(kubeClient)
 	defer close(stopChan)
 
 	for name, tc := range testcases {
@@ -1085,7 +1085,7 @@ func TestTopologyAggregation(t *testing.T) {
 
 			kubeClient := fakeclientset.NewSimpleClientset(nodes, csiNodes)
 
-			_, csiNodeLister, nodeLister, _, stopChan := listers(kubeClient)
+			_, csiNodeLister, nodeLister, stopChan := listers(kubeClient)
 			defer close(stopChan)
 
 			var selectedNode *v1.Node
@@ -1323,6 +1323,73 @@ func TestPreferredTopologies(t *testing.T) {
 			},
 			expectError: true,
 		},
+		"allowedTopologies is subset of selected node's topology": {
+			allowedTopologies: []v1.TopologySelectorTerm{
+				{
+					MatchLabelExpressions: []v1.TopologySelectorLabelRequirement{
+						{
+							Key:    "com.example.csi/disk",
+							Values: []string{"ssd"},
+						},
+						{
+							Key:    "com.example.csi/zone",
+							Values: []string{"zone1"},
+						},
+					},
+				},
+			},
+			nodeLabels: []map[string]string{
+				{"com.example.csi/zone": "zone1", "com.example.csi/rack": "rack1", "com.example.csi/disk": "ssd"},
+				{"com.example.csi/zone": "zone2", "com.example.csi/rack": "rack2", "com.example.csi/disk": "ssd"},
+				{"com.example.csi/zone": "zone2", "com.example.csi/rack": "rack1", "com.example.csi/disk": "nvme"},
+			},
+			topologyKeys: []map[string][]string{
+				{testDriverName: []string{"com.example.csi/zone", "com.example.csi/rack", "com.example.csi/disk"}},
+				{testDriverName: []string{"com.example.csi/zone", "com.example.csi/rack", "com.example.csi/disk"}},
+				{testDriverName: []string{"com.example.csi/zone", "com.example.csi/rack", "com.example.csi/disk"}},
+			},
+			expectError: false,
+			expectedPreferred: []*csi.Topology{
+				{
+					Segments: map[string]string{
+						"com.example.csi/disk": "ssd",
+						"com.example.csi/zone": "zone1",
+					},
+				},
+			},
+			expectedStrictPreferred: []*csi.Topology{
+				{
+					Segments: map[string]string{
+						"com.example.csi/disk": "ssd",
+						"com.example.csi/rack": "rack1",
+						"com.example.csi/zone": "zone1",
+					},
+				},
+			},
+		},
+		"allowedTopologies is not the subset of selected node's topology": {
+			allowedTopologies: []v1.TopologySelectorTerm{
+				{
+					MatchLabelExpressions: []v1.TopologySelectorLabelRequirement{
+						{
+							Key:    "com.example.csi/zone",
+							Values: []string{"zone1"},
+						},
+					},
+				},
+			},
+			nodeLabels: []map[string]string{
+				{"com.example.csi/zone": "zone10", "com.example.csi/rack": "rack1"},
+				{"com.example.csi/zone": "zone11", "com.example.csi/rack": "rack2"},
+				{"com.example.csi/zone": "zone12", "com.example.csi/rack": "rack1"},
+			},
+			topologyKeys: []map[string][]string{
+				{testDriverName: []string{"com.example.csi/zone", "com.example.csi/rack"}},
+				{testDriverName: []string{"com.example.csi/zone", "com.example.csi/rack"}},
+				{testDriverName: []string{"com.example.csi/zone", "com.example.csi/rack"}},
+			},
+			expectError: true,
+		},
 	}
 
 	for name, tc := range testcases {
@@ -1338,7 +1405,7 @@ func TestPreferredTopologies(t *testing.T) {
 			kubeClient := fakeclientset.NewSimpleClientset(nodes, csiNodes)
 			selectedNode := &nodes.Items[0]
 
-			_, csiNodeLister, nodeLister, _, stopChan := listers(kubeClient)
+			_, csiNodeLister, nodeLister, stopChan := listers(kubeClient)
 			defer close(stopChan)
 
 			requirements, err := GenerateAccessibilityRequirements(
@@ -1552,19 +1619,13 @@ func requisiteEqual(t1, t2 []*csi.Topology) bool {
 	return unchecked.Len() == 0
 }
 
-func listers(kubeClient *fakeclientset.Clientset) (
-	storagelistersv1.StorageClassLister,
-	storagelistersv1beta1.CSINodeLister,
-	corelisters.NodeLister,
-	corelisters.PersistentVolumeClaimLister,
-	chan struct{}) {
+func listers(kubeClient *fakeclientset.Clientset) (storagelistersv1.StorageClassLister, storagelistersv1beta1.CSINodeLister, corelisters.NodeLister, chan struct{}) {
 	factory := informers.NewSharedInformerFactory(kubeClient, ResyncPeriodOfCsiNodeInformer)
 	stopChan := make(chan struct{})
 	scLister := factory.Storage().V1().StorageClasses().Lister()
 	csiNodeLister := factory.Storage().V1beta1().CSINodes().Lister()
 	nodeLister := factory.Core().V1().Nodes().Lister()
-	claimLister := factory.Core().V1().PersistentVolumeClaims().Lister()
 	factory.Start(stopChan)
 	factory.WaitForCacheSync(stopChan)
-	return scLister, csiNodeLister, nodeLister, claimLister, stopChan
+	return scLister, csiNodeLister, nodeLister, stopChan
 }