Address code review comments

Signed-off-by: Deep Debroy <[email protected]>

pkg/controller/controller.go

@@ -329,15 +329,15 @@ func getVolumeCapability(
 }
 
 func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.PersistentVolume, error) {
-	if options.PVC.Spec.Selector != nil {
-		return nil, fmt.Errorf("claim Selector is not supported")
-	}
-
 	createVolumeRequestParameters := options.Parameters
 	migratedVolume := false
 	if p.supportsMigrationFromInTreePluginName != "" {
 		storageClassName := options.PVC.Spec.StorageClassName
-		//TODO(https://github.com/kubernetes-csi/external-provisioner/issues/256): use informers
+		// TODO(https://github.com/kubernetes-csi/external-provisioner/issues/256): use informers
+		// NOTE: we cannot depend on PVC.Annotations[volume.beta.kubernetes.io/storage-provisioner] to get
+		// the in-tree provisioner name in case of CSI migration scenarios. The annotation will be
+		// set to the CSI provisioner name by PV controller for migration scenarios
+		// so that external provisioner can correctly pick up the PVC pointing to an in-tree plugin
 		storageClass, err := p.client.StorageV1().StorageClasses().Get(*storageClassName, metav1.GetOptions{})
 		if err != nil {
 			return nil, fmt.Errorf("failed to get storage class named %s: %v", *storageClassName, err)
@@ -548,6 +548,8 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 	if migratedVolume {
 		pv, err = csitranslationlib.TranslateCSIPVToInTree(pv)
 		if err != nil {
+			klog.Warningf("failed to translate CSI PV to in-tree due to: %v. Deleting provisioned PV", err)
+			p.Delete(pv)
 			return nil, err
 		}
 	}