diff --git a/pkg/volume/testing/testing.go b/pkg/volume/testing/testing.go
index e6e090a4ca..7edd8d0f7f 100644
--- a/pkg/volume/testing/testing.go
+++ b/pkg/volume/testing/testing.go
@@ -46,6 +46,10 @@ import (
 	"k8s.io/kubernetes/pkg/volume/util/volumepathhandler"
 )
 
+// A hook specified in storage class to indicate it's provisioning
+// is expected to fail.
+const ExpectProvisionFailureKey = "expect-provision-failure"
+
 // fakeVolumeHost is useful for testing volume plugins.
 type fakeVolumeHost struct {
 	rootDir    string
@@ -787,6 +791,12 @@ type FakeProvisioner struct {
 }
 
 func (fc *FakeProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
+	// Add provision failure hook
+	if fc.Options.Parameters != nil {
+		if _, ok := fc.Options.Parameters[ExpectProvisionFailureKey]; ok {
+			return nil, fmt.Errorf("expected error")
+		}
+	}
 	fullpath := fmt.Sprintf("/tmp/hostpath_pv/%s", uuid.NewUUID())
 
 	pv := &v1.PersistentVolume{
diff --git a/test/integration/scheduler/volume_binding_test.go b/test/integration/scheduler/volume_binding_test.go
index 9bb20f5adf..0608dc25ea 100644
--- a/test/integration/scheduler/volume_binding_test.go
+++ b/test/integration/scheduler/volume_binding_test.go
@@ -61,13 +61,14 @@ var (
 	modeWait      = storagev1.VolumeBindingWaitForFirstConsumer
 	modeImmediate = storagev1.VolumeBindingImmediate
 
-	classWait      = "wait"
-	classImmediate = "immediate"
-	classDynamic   = "dynamic"
+	classWait         = "wait"
+	classImmediate    = "immediate"
+	classDynamic      = "dynamic"
+	classTopoMismatch = "topomismatch"
 
-	sharedClasses = map[storagev1.VolumeBindingMode]*storagev1.StorageClass{
-		modeImmediate: makeStorageClass(classImmediate, &modeImmediate),
-		modeWait:      makeStorageClass(classWait, &modeWait),
+	sharedClasses = map[string]*storagev1.StorageClass{
+		classImmediate: makeStorageClass(classImmediate, &modeImmediate),
+		classWait:      makeStorageClass(classWait, &modeWait),
 	}
 )
 
@@ -82,14 +83,14 @@ const (
 
 type testPV struct {
 	name        string
-	scMode      storagev1.VolumeBindingMode
+	scName      string
 	preboundPVC string
 	node        string
 }
 
 type testPVC struct {
 	name       string
-	scMode     storagev1.VolumeBindingMode
+	scName     string
 	preboundPV string
 }
 
@@ -112,79 +113,79 @@ func TestVolumeBinding(t *testing.T) {
 	}{
 		"immediate can bind": {
 			pod:  makePod("pod-i-canbind", config.ns, []string{"pvc-i-canbind"}),
-			pvs:  []*testPV{{"pv-i-canbind", modeImmediate, "", node1}},
-			pvcs: []*testPVC{{"pvc-i-canbind", modeImmediate, ""}},
+			pvs:  []*testPV{{"pv-i-canbind", classImmediate, "", node1}},
+			pvcs: []*testPVC{{"pvc-i-canbind", classImmediate, ""}},
 		},
 		"immediate cannot bind": {
 			pod:         makePod("pod-i-cannotbind", config.ns, []string{"pvc-i-cannotbind"}),
-			unboundPvcs: []*testPVC{{"pvc-i-cannotbind", modeImmediate, ""}},
+			unboundPvcs: []*testPVC{{"pvc-i-cannotbind", classImmediate, ""}},
 			shouldFail:  true,
 		},
 		"immediate pvc prebound": {
 			pod:  makePod("pod-i-pvc-prebound", config.ns, []string{"pvc-i-prebound"}),
-			pvs:  []*testPV{{"pv-i-pvc-prebound", modeImmediate, "", node1}},
-			pvcs: []*testPVC{{"pvc-i-prebound", modeImmediate, "pv-i-pvc-prebound"}},
+			pvs:  []*testPV{{"pv-i-pvc-prebound", classImmediate, "", node1}},
+			pvcs: []*testPVC{{"pvc-i-prebound", classImmediate, "pv-i-pvc-prebound"}},
 		},
 		"immediate pv prebound": {
 			pod:  makePod("pod-i-pv-prebound", config.ns, []string{"pvc-i-pv-prebound"}),
-			pvs:  []*testPV{{"pv-i-prebound", modeImmediate, "pvc-i-pv-prebound", node1}},
-			pvcs: []*testPVC{{"pvc-i-pv-prebound", modeImmediate, ""}},
+			pvs:  []*testPV{{"pv-i-prebound", classImmediate, "pvc-i-pv-prebound", node1}},
+			pvcs: []*testPVC{{"pvc-i-pv-prebound", classImmediate, ""}},
 		},
 		"wait can bind": {
 			pod:  makePod("pod-w-canbind", config.ns, []string{"pvc-w-canbind"}),
-			pvs:  []*testPV{{"pv-w-canbind", modeWait, "", node1}},
-			pvcs: []*testPVC{{"pvc-w-canbind", modeWait, ""}},
+			pvs:  []*testPV{{"pv-w-canbind", classWait, "", node1}},
+			pvcs: []*testPVC{{"pvc-w-canbind", classWait, ""}},
 		},
 		"wait cannot bind": {
 			pod:         makePod("pod-w-cannotbind", config.ns, []string{"pvc-w-cannotbind"}),
-			unboundPvcs: []*testPVC{{"pvc-w-cannotbind", modeWait, ""}},
+			unboundPvcs: []*testPVC{{"pvc-w-cannotbind", classWait, ""}},
 			shouldFail:  true,
 		},
 		"wait pvc prebound": {
 			pod:  makePod("pod-w-pvc-prebound", config.ns, []string{"pvc-w-prebound"}),
-			pvs:  []*testPV{{"pv-w-pvc-prebound", modeWait, "", node1}},
-			pvcs: []*testPVC{{"pvc-w-prebound", modeWait, "pv-w-pvc-prebound"}},
+			pvs:  []*testPV{{"pv-w-pvc-prebound", classWait, "", node1}},
+			pvcs: []*testPVC{{"pvc-w-prebound", classWait, "pv-w-pvc-prebound"}},
 		},
 		"wait pv prebound": {
 			pod:  makePod("pod-w-pv-prebound", config.ns, []string{"pvc-w-pv-prebound"}),
-			pvs:  []*testPV{{"pv-w-prebound", modeWait, "pvc-w-pv-prebound", node1}},
-			pvcs: []*testPVC{{"pvc-w-pv-prebound", modeWait, ""}},
+			pvs:  []*testPV{{"pv-w-prebound", classWait, "pvc-w-pv-prebound", node1}},
+			pvcs: []*testPVC{{"pvc-w-pv-prebound", classWait, ""}},
 		},
 		"wait can bind two": {
 			pod: makePod("pod-w-canbind-2", config.ns, []string{"pvc-w-canbind-2", "pvc-w-canbind-3"}),
 			pvs: []*testPV{
-				{"pv-w-canbind-2", modeWait, "", node2},
-				{"pv-w-canbind-3", modeWait, "", node2},
+				{"pv-w-canbind-2", classWait, "", node2},
+				{"pv-w-canbind-3", classWait, "", node2},
 			},
 			pvcs: []*testPVC{
-				{"pvc-w-canbind-2", modeWait, ""},
-				{"pvc-w-canbind-3", modeWait, ""},
+				{"pvc-w-canbind-2", classWait, ""},
+				{"pvc-w-canbind-3", classWait, ""},
 			},
 			unboundPvs: []*testPV{
-				{"pv-w-canbind-5", modeWait, "", node1},
+				{"pv-w-canbind-5", classWait, "", node1},
 			},
 		},
 		"wait cannot bind two": {
 			pod: makePod("pod-w-cannotbind-2", config.ns, []string{"pvc-w-cannotbind-1", "pvc-w-cannotbind-2"}),
 			unboundPvcs: []*testPVC{
-				{"pvc-w-cannotbind-1", modeWait, ""},
-				{"pvc-w-cannotbind-2", modeWait, ""},
+				{"pvc-w-cannotbind-1", classWait, ""},
+				{"pvc-w-cannotbind-2", classWait, ""},
 			},
 			unboundPvs: []*testPV{
-				{"pv-w-cannotbind-1", modeWait, "", node2},
-				{"pv-w-cannotbind-2", modeWait, "", node1},
+				{"pv-w-cannotbind-1", classWait, "", node2},
+				{"pv-w-cannotbind-2", classWait, "", node1},
 			},
 			shouldFail: true,
 		},
 		"mix immediate and wait": {
 			pod: makePod("pod-mix-bound", config.ns, []string{"pvc-w-canbind-4", "pvc-i-canbind-2"}),
 			pvs: []*testPV{
-				{"pv-w-canbind-4", modeWait, "", node1},
-				{"pv-i-canbind-2", modeImmediate, "", node1},
+				{"pv-w-canbind-4", classWait, "", node1},
+				{"pv-i-canbind-2", classImmediate, "", node1},
 			},
 			pvcs: []*testPVC{
-				{"pvc-w-canbind-4", modeWait, ""},
-				{"pvc-i-canbind-2", modeImmediate, ""},
+				{"pvc-w-canbind-4", classWait, ""},
+				{"pvc-i-canbind-2", classImmediate, ""},
 			},
 		},
 	}
@@ -194,9 +195,9 @@ func TestVolumeBinding(t *testing.T) {
 
 		// Create two StorageClasses
 		suffix := rand.String(4)
-		classes := map[storagev1.VolumeBindingMode]*storagev1.StorageClass{}
-		classes[modeImmediate] = makeStorageClass(fmt.Sprintf("immediate-%v", suffix), &modeImmediate)
-		classes[modeWait] = makeStorageClass(fmt.Sprintf("wait-%v", suffix), &modeWait)
+		classes := map[string]*storagev1.StorageClass{}
+		classes[classImmediate] = makeStorageClass(fmt.Sprintf("immediate-%v", suffix), &modeImmediate)
+		classes[classWait] = makeStorageClass(fmt.Sprintf("wait-%v", suffix), &modeWait)
 		for _, sc := range classes {
 			if _, err := config.client.StorageV1().StorageClasses().Create(sc); err != nil {
 				t.Fatalf("Failed to create StorageClass %q: %v", sc.Name, err)
@@ -205,14 +206,14 @@ func TestVolumeBinding(t *testing.T) {
 
 		// Create PVs
 		for _, pvConfig := range test.pvs {
-			pv := makePV(pvConfig.name, classes[pvConfig.scMode].Name, pvConfig.preboundPVC, config.ns, pvConfig.node)
+			pv := makePV(pvConfig.name, classes[pvConfig.scName].Name, pvConfig.preboundPVC, config.ns, pvConfig.node)
 			if _, err := config.client.CoreV1().PersistentVolumes().Create(pv); err != nil {
 				t.Fatalf("Failed to create PersistentVolume %q: %v", pv.Name, err)
 			}
 		}
 
 		for _, pvConfig := range test.unboundPvs {
-			pv := makePV(pvConfig.name, classes[pvConfig.scMode].Name, pvConfig.preboundPVC, config.ns, pvConfig.node)
+			pv := makePV(pvConfig.name, classes[pvConfig.scName].Name, pvConfig.preboundPVC, config.ns, pvConfig.node)
 			if _, err := config.client.CoreV1().PersistentVolumes().Create(pv); err != nil {
 				t.Fatalf("Failed to create PersistentVolume %q: %v", pv.Name, err)
 			}
@@ -220,13 +221,13 @@ func TestVolumeBinding(t *testing.T) {
 
 		// Create PVCs
 		for _, pvcConfig := range test.pvcs {
-			pvc := makePVC(pvcConfig.name, config.ns, &classes[pvcConfig.scMode].Name, pvcConfig.preboundPV)
+			pvc := makePVC(pvcConfig.name, config.ns, &classes[pvcConfig.scName].Name, pvcConfig.preboundPV)
 			if _, err := config.client.CoreV1().PersistentVolumeClaims(config.ns).Create(pvc); err != nil {
 				t.Fatalf("Failed to create PersistentVolumeClaim %q: %v", pvc.Name, err)
 			}
 		}
 		for _, pvcConfig := range test.unboundPvcs {
-			pvc := makePVC(pvcConfig.name, config.ns, &classes[pvcConfig.scMode].Name, pvcConfig.preboundPV)
+			pvc := makePVC(pvcConfig.name, config.ns, &classes[pvcConfig.scName].Name, pvcConfig.preboundPV)
 			if _, err := config.client.CoreV1().PersistentVolumeClaims(config.ns).Create(pvc); err != nil {
 				t.Fatalf("Failed to create PersistentVolumeClaim %q: %v", pvc.Name, err)
 			}
@@ -248,10 +249,10 @@ func TestVolumeBinding(t *testing.T) {
 
 		// Validate PVC/PV binding
 		for _, pvc := range test.pvcs {
-			validatePVCPhase(t, config.client, pvc.name, config.ns, v1.ClaimBound)
+			validatePVCPhase(t, config.client, pvc.name, config.ns, v1.ClaimBound, false)
 		}
 		for _, pvc := range test.unboundPvcs {
-			validatePVCPhase(t, config.client, pvc.name, config.ns, v1.ClaimPending)
+			validatePVCPhase(t, config.client, pvc.name, config.ns, v1.ClaimPending, false)
 		}
 		for _, pv := range test.pvs {
 			validatePVPhase(t, config.client, pv.name, v1.VolumeBound)
@@ -289,7 +290,7 @@ func TestVolumeBindingRescheduling(t *testing.T) {
 				{"pvc-reschedule-onclassadd-dynamic", "", ""},
 			},
 			trigger: func(config *testConfig) {
-				sc := makeDynamicProvisionerStorageClass(storageClassName, &modeWait)
+				sc := makeDynamicProvisionerStorageClass(storageClassName, &modeWait, nil)
 				if _, err := config.client.StorageV1().StorageClasses().Create(sc); err != nil {
 					t.Fatalf("Failed to create StorageClass %q: %v", sc.Name, err)
 				}
@@ -319,7 +320,7 @@ func TestVolumeBindingRescheduling(t *testing.T) {
 			pvs: []*testPV{
 				{
 					name:   "pv-reschedule-onpvcadd",
-					scMode: modeWait,
+					scName: classWait,
 					node:   node1,
 				},
 			},
@@ -350,7 +351,7 @@ func TestVolumeBindingRescheduling(t *testing.T) {
 
 		// Create PVs
 		for _, pvConfig := range test.pvs {
-			pv := makePV(pvConfig.name, sharedClasses[pvConfig.scMode].Name, pvConfig.preboundPVC, config.ns, pvConfig.node)
+			pv := makePV(pvConfig.name, sharedClasses[pvConfig.scName].Name, pvConfig.preboundPVC, config.ns, pvConfig.node)
 			if _, err := config.client.CoreV1().PersistentVolumes().Create(pv); err != nil {
 				t.Fatalf("Failed to create PersistentVolume %q: %v", pv.Name, err)
 			}
@@ -430,7 +431,7 @@ func testVolumeBindingStress(t *testing.T, schedulerResyncPeriod time.Duration,
 	scName := &classWait
 	if dynamic {
 		scName = &classDynamic
-		sc := makeDynamicProvisionerStorageClass(*scName, &modeWait)
+		sc := makeDynamicProvisionerStorageClass(*scName, &modeWait, nil)
 		if _, err := config.client.StorageV1().StorageClasses().Create(sc); err != nil {
 			t.Fatalf("Failed to create StorageClass %q: %v", sc.Name, err)
 		}
@@ -482,7 +483,7 @@ func testVolumeBindingStress(t *testing.T, schedulerResyncPeriod time.Duration,
 
 	// Validate PVC/PV binding
 	for _, pvc := range pvcs {
-		validatePVCPhase(t, config.client, pvc.Name, config.ns, v1.ClaimBound)
+		validatePVCPhase(t, config.client, pvc.Name, config.ns, v1.ClaimBound, dynamic)
 	}
 	for _, pv := range pvs {
 		validatePVPhase(t, config.client, pv.Name, v1.VolumeBound)
@@ -595,7 +596,7 @@ func testVolumeBindingWithAffinity(t *testing.T, anti bool, numNodes, numPods, n
 
 	// Validate PVC binding
 	for _, pvc := range pvcs {
-		validatePVCPhase(t, config.client, pvc.Name, config.ns, v1.ClaimBound)
+		validatePVCPhase(t, config.client, pvc.Name, config.ns, v1.ClaimBound, false)
 	}
 }
 
@@ -682,6 +683,222 @@ func TestPVAffinityConflict(t *testing.T) {
 	}
 }
 
+func TestVolumeProvision(t *testing.T) {
+	features := map[string]bool{
+		"VolumeScheduling":       true,
+		"PersistentLocalVolumes": true,
+	}
+	config := setupCluster(t, "volume-scheduling", 1, features, 0, 0, false)
+	defer config.teardown()
+
+	cases := map[string]struct {
+		pod             *v1.Pod
+		pvs             []*testPV
+		boundPvcs       []*testPVC
+		provisionedPvcs []*testPVC
+		// Create these, but they should not be bound in the end
+		unboundPvcs []*testPVC
+		shouldFail  bool
+	}{
+		"wait provisioned": {
+			pod:             makePod("pod-pvc-canprovision", config.ns, []string{"pvc-canprovision"}),
+			provisionedPvcs: []*testPVC{{"pvc-canprovision", classWait, ""}},
+		},
+		"topolgy unsatisfied": {
+			pod:         makePod("pod-pvc-topomismatch", config.ns, []string{"pvc-topomismatch"}),
+			unboundPvcs: []*testPVC{{"pvc-topomismatch", classTopoMismatch, ""}},
+			shouldFail:  true,
+		},
+		"wait one bound, one provisioned": {
+			pod:             makePod("pod-pvc-canbind-or-provision", config.ns, []string{"pvc-w-canbind", "pvc-canprovision"}),
+			pvs:             []*testPV{{"pv-w-canbind", classWait, "", node1}},
+			boundPvcs:       []*testPVC{{"pvc-w-canbind", classWait, ""}},
+			provisionedPvcs: []*testPVC{{"pvc-canprovision", classWait, ""}},
+		},
+		"one immediate pvc prebound, one wait provisioned": {
+			pod:             makePod("pod-i-pvc-prebound-w-provisioned", config.ns, []string{"pvc-i-prebound", "pvc-canprovision"}),
+			pvs:             []*testPV{{"pv-i-pvc-prebound", classImmediate, "", node1}},
+			boundPvcs:       []*testPVC{{"pvc-i-prebound", classImmediate, "pv-i-pvc-prebound"}},
+			provisionedPvcs: []*testPVC{{"pvc-canprovision", classWait, ""}},
+		},
+		"one immediate pv prebound, one wait provisioned": {
+			pod:             makePod("pod-i-pv-prebound-w-provisioned", config.ns, []string{"pvc-i-pv-prebound", "pvc-canprovision"}),
+			pvs:             []*testPV{{"pv-i-prebound", classImmediate, "pvc-i-pv-prebound", node1}},
+			boundPvcs:       []*testPVC{{"pvc-i-pv-prebound", classImmediate, ""}},
+			provisionedPvcs: []*testPVC{{"pvc-canprovision", classWait, ""}},
+		},
+		"wait one pv prebound, one provisioned": {
+			pod:             makePod("pod-w-pv-prebound-w-provisioned", config.ns, []string{"pvc-w-pv-prebound", "pvc-canprovision"}),
+			pvs:             []*testPV{{"pv-w-prebound", classWait, "pvc-w-pv-prebound", node1}},
+			boundPvcs:       []*testPVC{{"pvc-w-pv-prebound", classWait, ""}},
+			provisionedPvcs: []*testPVC{{"pvc-canprovision", classWait, ""}},
+		},
+		"immediate provisioned by controller": {
+			pod: makePod("pod-i-unbound", config.ns, []string{"pvc-controller-provisioned"}),
+			// A pvc of immediate binding mode is expected to be provisioned by controller,
+			// we treat it as "bound" here because it is supposed to be in same state
+			// with bound claims, i.e. in bound status and has no selectedNode annotation.
+			boundPvcs: []*testPVC{{"pvc-controller-provisioned", classImmediate, ""}},
+		},
+	}
+
+	for name, test := range cases {
+		glog.Infof("Running test %v", name)
+
+		// Create StorageClasses
+		suffix := rand.String(4)
+		classes := map[string]*storagev1.StorageClass{}
+		classes[classImmediate] = makeDynamicProvisionerStorageClass(fmt.Sprintf("immediate-%v", suffix), &modeImmediate, nil)
+		classes[classWait] = makeDynamicProvisionerStorageClass(fmt.Sprintf("wait-%v", suffix), &modeWait, nil)
+		topo := []v1.TopologySelectorTerm{
+			{
+				MatchLabelExpressions: []v1.TopologySelectorLabelRequirement{
+					{
+						Key:    nodeAffinityLabelKey,
+						Values: []string{node2},
+					},
+				},
+			},
+		}
+		classes[classTopoMismatch] = makeDynamicProvisionerStorageClass(fmt.Sprintf("topomismatch-%v", suffix), &modeWait, topo)
+		for _, sc := range classes {
+			if _, err := config.client.StorageV1().StorageClasses().Create(sc); err != nil {
+				t.Fatalf("Failed to create StorageClass %q: %v", sc.Name, err)
+			}
+		}
+		// Create PVs
+		for _, pvConfig := range test.pvs {
+			pv := makePV(pvConfig.name, classes[pvConfig.scName].Name, pvConfig.preboundPVC, config.ns, pvConfig.node)
+			if _, err := config.client.CoreV1().PersistentVolumes().Create(pv); err != nil {
+				t.Fatalf("Failed to create PersistentVolume %q: %v", pv.Name, err)
+			}
+		}
+
+		// Create PVCs
+		for _, pvcConfig := range test.boundPvcs {
+			pvc := makePVC(pvcConfig.name, config.ns, &classes[pvcConfig.scName].Name, pvcConfig.preboundPV)
+			if _, err := config.client.CoreV1().PersistentVolumeClaims(config.ns).Create(pvc); err != nil {
+				t.Fatalf("Failed to create PersistentVolumeClaim %q: %v", pvc.Name, err)
+			}
+		}
+		for _, pvcConfig := range test.unboundPvcs {
+			pvc := makePVC(pvcConfig.name, config.ns, &classes[pvcConfig.scName].Name, pvcConfig.preboundPV)
+			if _, err := config.client.CoreV1().PersistentVolumeClaims(config.ns).Create(pvc); err != nil {
+				t.Fatalf("Failed to create PersistentVolumeClaim %q: %v", pvc.Name, err)
+			}
+		}
+		for _, pvcConfig := range test.provisionedPvcs {
+			pvc := makePVC(pvcConfig.name, config.ns, &classes[pvcConfig.scName].Name, pvcConfig.preboundPV)
+			if _, err := config.client.CoreV1().PersistentVolumeClaims(config.ns).Create(pvc); err != nil {
+				t.Fatalf("Failed to create PersistentVolumeClaim %q: %v", pvc.Name, err)
+			}
+		}
+
+		// Create Pod
+		if _, err := config.client.CoreV1().Pods(config.ns).Create(test.pod); err != nil {
+			t.Fatalf("Failed to create Pod %q: %v", test.pod.Name, err)
+		}
+		if test.shouldFail {
+			if err := waitForPodUnschedulable(config.client, test.pod); err != nil {
+				t.Errorf("Pod %q was not unschedulable: %v", test.pod.Name, err)
+			}
+		} else {
+			if err := waitForPodToSchedule(config.client, test.pod); err != nil {
+				t.Errorf("Failed to schedule Pod %q: %v", test.pod.Name, err)
+			}
+		}
+
+		// Validate PVC/PV binding
+		for _, pvc := range test.boundPvcs {
+			validatePVCPhase(t, config.client, pvc.name, config.ns, v1.ClaimBound, false)
+		}
+		for _, pvc := range test.unboundPvcs {
+			validatePVCPhase(t, config.client, pvc.name, config.ns, v1.ClaimPending, false)
+		}
+		for _, pvc := range test.provisionedPvcs {
+			validatePVCPhase(t, config.client, pvc.name, config.ns, v1.ClaimBound, true)
+		}
+		for _, pv := range test.pvs {
+			validatePVPhase(t, config.client, pv.name, v1.VolumeBound)
+		}
+
+		// Force delete objects, but they still may not be immediately removed
+		deleteTestObjects(config.client, config.ns, deleteOption)
+	}
+}
+
+// TestRescheduleProvisioning validate that PV controller will remove
+// selectedNode annotation from a claim to reschedule volume provision
+// on provision failure.
+func TestRescheduleProvisioning(t *testing.T) {
+	features := map[string]bool{
+		"VolumeScheduling": true,
+	}
+	oldFeatures := make(map[string]bool, len(features))
+	for feature := range features {
+		oldFeatures[feature] = utilfeature.DefaultFeatureGate.Enabled(utilfeature.Feature(feature))
+	}
+	// Set feature gates
+	utilfeature.DefaultFeatureGate.SetFromMap(features)
+	controllerCh := make(chan struct{})
+
+	context := initTestMaster(t, "reschedule-volume-provision", nil)
+
+	clientset := context.clientSet
+	ns := context.ns.Name
+
+	defer func() {
+		close(controllerCh)
+		deleteTestObjects(clientset, ns, nil)
+		context.clientSet.CoreV1().Nodes().DeleteCollection(nil, metav1.ListOptions{})
+		context.closeFn()
+		// Restore feature gates
+		utilfeature.DefaultFeatureGate.SetFromMap(oldFeatures)
+	}()
+
+	ctrl, informerFactory, err := initPVController(context, 0)
+	if err != nil {
+		t.Fatalf("Failed to create PV controller: %v", err)
+	}
+
+	// Prepare node and storage class.
+	testNode := makeNode(0)
+	if _, err := clientset.CoreV1().Nodes().Create(testNode); err != nil {
+		t.Fatalf("Failed to create Node %q: %v", testNode.Name, err)
+	}
+	scName := "fail-provision"
+	sc := makeDynamicProvisionerStorageClass(scName, &modeWait, nil)
+	// Expect the storage class fail to provision.
+	sc.Parameters[volumetest.ExpectProvisionFailureKey] = ""
+	if _, err := clientset.StorageV1().StorageClasses().Create(sc); err != nil {
+		t.Fatalf("Failed to create StorageClass %q: %v", sc.Name, err)
+	}
+
+	// Create a pvc with selected node annotation.
+	pvcName := "pvc-fail-to-provision"
+	pvc := makePVC(pvcName, ns, &scName, "")
+	pvc.Annotations = map[string]string{"volume.kubernetes.io/selected-node": node1}
+	pvc, err = clientset.CoreV1().PersistentVolumeClaims(ns).Create(pvc)
+	if err != nil {
+		t.Fatalf("Failed to create PersistentVolumeClaim %q: %v", pvc.Name, err)
+	}
+	// Validate selectedNode annotation exists on created claim.
+	selectedNodeAnn, exist := pvc.Annotations["volume.kubernetes.io/selected-node"]
+	if !exist || selectedNodeAnn != node1 {
+		t.Fatalf("Created pvc is not annotated as expected")
+	}
+
+	// Start controller.
+	go ctrl.Run(controllerCh)
+	informerFactory.Start(controllerCh)
+	informerFactory.WaitForCacheSync(controllerCh)
+
+	// Validate that the annotation is removed by controller for provision reschedule.
+	if err := waitForProvisionAnn(clientset, pvc, false); err != nil {
+		t.Errorf("Expect to reschedule provision for PVC %v/%v, but still found selected-node annotation on it", ns, pvcName)
+	}
+}
+
 func setupCluster(t *testing.T, nsName string, numberOfNodes int, features map[string]bool, resyncPeriod time.Duration, provisionDelaySeconds int, disableEquivalenceCache bool) *testConfig {
 	oldFeatures := make(map[string]bool, len(features))
 	for feature := range features {
@@ -696,8 +913,49 @@ func setupCluster(t *testing.T, nsName string, numberOfNodes int, features map[s
 
 	clientset := context.clientSet
 	ns := context.ns.Name
+
+	ctrl, informerFactory, err := initPVController(context, provisionDelaySeconds)
+	if err != nil {
+		t.Fatalf("Failed to create PV controller: %v", err)
+	}
+	go ctrl.Run(controllerCh)
+	// Start informer factory after all controllers are configured and running.
+	informerFactory.Start(controllerCh)
+	informerFactory.WaitForCacheSync(controllerCh)
+
+	// Create shared objects
+	// Create nodes
+	for i := 0; i < numberOfNodes; i++ {
+		testNode := makeNode(i)
+		if _, err := clientset.CoreV1().Nodes().Create(testNode); err != nil {
+			t.Fatalf("Failed to create Node %q: %v", testNode.Name, err)
+		}
+	}
+
+	// Create SCs
+	for _, sc := range sharedClasses {
+		if _, err := clientset.StorageV1().StorageClasses().Create(sc); err != nil {
+			t.Fatalf("Failed to create StorageClass %q: %v", sc.Name, err)
+		}
+	}
+
+	return &testConfig{
+		client: clientset,
+		ns:     ns,
+		stop:   controllerCh,
+		teardown: func() {
+			deleteTestObjects(clientset, ns, nil)
+			cleanupTest(t, context)
+			// Restore feature gates
+			utilfeature.DefaultFeatureGate.SetFromMap(oldFeatures)
+		},
+	}
+}
+
+func initPVController(context *TestContext, provisionDelaySeconds int) (*persistentvolume.PersistentVolumeController, informers.SharedInformerFactory, error) {
+	clientset := context.clientSet
 	// Informers factory for controllers, we disable resync period for testing.
-	informerFactory := informers.NewSharedInformerFactory(context.clientSet, 0)
+	informerFactory := informers.NewSharedInformerFactory(clientset, 0)
 
 	// Start PV controller for volume binding.
 	host := volumetest.NewFakeVolumeHost("/tmp/fake", nil, nil)
@@ -730,61 +988,13 @@ func setupCluster(t *testing.T, nsName string, numberOfNodes int, features map[s
 		NodeInformer:              informerFactory.Core().V1().Nodes(),
 		EnableDynamicProvisioning: true,
 	}
+
 	ctrl, err := persistentvolume.NewController(params)
 	if err != nil {
-		t.Fatalf("Failed to create PV controller: %v", err)
-	}
-	go ctrl.Run(controllerCh)
-	// Start informer factory after all controllers are configured and running.
-	informerFactory.Start(controllerCh)
-	informerFactory.WaitForCacheSync(controllerCh)
-
-	// Create shared objects
-	// Create nodes
-	for i := 0; i < numberOfNodes; i++ {
-		testNode := &v1.Node{
-			ObjectMeta: metav1.ObjectMeta{
-				Name:   fmt.Sprintf("node-%d", i+1),
-				Labels: map[string]string{nodeAffinityLabelKey: fmt.Sprintf("node-%d", i+1)},
-			},
-			Spec: v1.NodeSpec{Unschedulable: false},
-			Status: v1.NodeStatus{
-				Capacity: v1.ResourceList{
-					v1.ResourcePods: *resource.NewQuantity(podLimit, resource.DecimalSI),
-				},
-				Conditions: []v1.NodeCondition{
-					{
-						Type:              v1.NodeReady,
-						Status:            v1.ConditionTrue,
-						Reason:            fmt.Sprintf("schedulable condition"),
-						LastHeartbeatTime: metav1.Time{Time: time.Now()},
-					},
-				},
-			},
-		}
-		if _, err := clientset.CoreV1().Nodes().Create(testNode); err != nil {
-			t.Fatalf("Failed to create Node %q: %v", testNode.Name, err)
-		}
+		return nil, nil, err
 	}
 
-	// Create SCs
-	for _, sc := range sharedClasses {
-		if _, err := clientset.StorageV1().StorageClasses().Create(sc); err != nil {
-			t.Fatalf("Failed to create StorageClass %q: %v", sc.Name, err)
-		}
-	}
-
-	return &testConfig{
-		client: clientset,
-		ns:     ns,
-		stop:   controllerCh,
-		teardown: func() {
-			deleteTestObjects(clientset, ns, nil)
-			cleanupTest(t, context)
-			// Restore feature gates
-			utilfeature.DefaultFeatureGate.SetFromMap(oldFeatures)
-		},
-	}
+	return ctrl, informerFactory, nil
 }
 
 func deleteTestObjects(client clientset.Interface, ns string, option *metav1.DeleteOptions) {
@@ -804,13 +1014,15 @@ func makeStorageClass(name string, mode *storagev1.VolumeBindingMode) *storagev1
 	}
 }
 
-func makeDynamicProvisionerStorageClass(name string, mode *storagev1.VolumeBindingMode) *storagev1.StorageClass {
+func makeDynamicProvisionerStorageClass(name string, mode *storagev1.VolumeBindingMode, allowedTopologies []v1.TopologySelectorTerm) *storagev1.StorageClass {
 	return &storagev1.StorageClass{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: name,
 		},
 		Provisioner:       provisionerPluginName,
 		VolumeBindingMode: mode,
+		AllowedTopologies: allowedTopologies,
+		Parameters:        map[string]string{},
 	}
 }
 
@@ -914,7 +1126,30 @@ func makePod(name, ns string, pvcs []string) *v1.Pod {
 	}
 }
 
-func validatePVCPhase(t *testing.T, client clientset.Interface, pvcName string, ns string, phase v1.PersistentVolumeClaimPhase) {
+func makeNode(index int) *v1.Node {
+	return &v1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   fmt.Sprintf("node-%d", index+1),
+			Labels: map[string]string{nodeAffinityLabelKey: fmt.Sprintf("node-%d", index+1)},
+		},
+		Spec: v1.NodeSpec{Unschedulable: false},
+		Status: v1.NodeStatus{
+			Capacity: v1.ResourceList{
+				v1.ResourcePods: *resource.NewQuantity(podLimit, resource.DecimalSI),
+			},
+			Conditions: []v1.NodeCondition{
+				{
+					Type:              v1.NodeReady,
+					Status:            v1.ConditionTrue,
+					Reason:            fmt.Sprintf("schedulable condition"),
+					LastHeartbeatTime: metav1.Time{Time: time.Now()},
+				},
+			},
+		},
+	}
+}
+
+func validatePVCPhase(t *testing.T, client clientset.Interface, pvcName string, ns string, phase v1.PersistentVolumeClaimPhase, isProvisioned bool) {
 	claim, err := client.CoreV1().PersistentVolumeClaims(ns).Get(pvcName, metav1.GetOptions{})
 	if err != nil {
 		t.Errorf("Failed to get PVC %v/%v: %v", ns, pvcName, err)
@@ -923,6 +1158,43 @@ func validatePVCPhase(t *testing.T, client clientset.Interface, pvcName string,
 	if claim.Status.Phase != phase {
 		t.Errorf("PVC %v/%v phase not %v, got %v", ns, pvcName, phase, claim.Status.Phase)
 	}
+
+	// Check whether the bound claim is provisioned/bound as expect.
+	if phase == v1.ClaimBound {
+		if err := validateProvisionAnn(claim, isProvisioned); err != nil {
+			t.Errorf("Provisoning annotaion on PVC %v/%v not bahaviors as expected: %v", ns, pvcName, err)
+		}
+	}
+}
+
+func validateProvisionAnn(claim *v1.PersistentVolumeClaim, volIsProvisioned bool) error {
+	selectedNode, provisionAnnoExist := claim.Annotations["volume.kubernetes.io/selected-node"]
+	if volIsProvisioned {
+		if !provisionAnnoExist {
+			return fmt.Errorf("PVC %v/%v expected to be provisioned, but no selected-node annotation found", claim.Namespace, claim.Name)
+		}
+		if selectedNode != node1 {
+			return fmt.Errorf("PVC %v/%v expected to be annotated as %v, but got %v", claim.Namespace, claim.Name, node1, selectedNode)
+		}
+	}
+	if !volIsProvisioned && provisionAnnoExist {
+		return fmt.Errorf("PVC %v/%v not expected to be provisioned, but found selected-node annotation", claim.Namespace, claim.Name)
+	}
+
+	return nil
+}
+
+func waitForProvisionAnn(client clientset.Interface, pvc *v1.PersistentVolumeClaim, annShouldExist bool) error {
+	return wait.Poll(time.Second, 30*time.Second, func() (bool, error) {
+		claim, err := client.CoreV1().PersistentVolumeClaims(pvc.Namespace).Get(pvc.Name, metav1.GetOptions{})
+		if err != nil {
+			return false, err
+		}
+		if err := validateProvisionAnn(claim, annShouldExist); err == nil {
+			return true, nil
+		}
+		return false, nil
+	})
 }
 
 func validatePVPhase(t *testing.T, client clientset.Interface, pvName string, phase v1.PersistentVolumePhase) {