diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
index 509ff63e6b..36d0b7da47 100644
--- a/pkg/controller/controller.go
+++ b/pkg/controller/controller.go
@@ -491,7 +491,11 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 		volumeAttributes[k] = v
 	}
 	respCap := rep.GetVolume().GetCapacityBytes()
-	if respCap < volSizeBytes {
+
+	//According to CSI spec CreateVolume should be able to return capacity = 0, which means it is unknown. for example NFS/FTP
+	if respCap == 0 {
+		klog.V(3).Info("response volume with size = 0")
+	} else if respCap < volSizeBytes {
 		capErr := fmt.Errorf("created volume capacity %v less than requested capacity %v", respCap, volSizeBytes)
 		delReq := &csi.DeleteVolumeRequest{
 			VolumeId: rep.GetVolume().GetVolumeId(),
diff --git a/pkg/controller/controller_test.go b/pkg/controller/controller_test.go
index 1b6ffffbfe..fd40bec156 100644
--- a/pkg/controller/controller_test.go
+++ b/pkg/controller/controller_test.go
@@ -634,6 +634,7 @@ type provisioningTestcase struct {
 	getSecretRefErr   bool
 	getCredentialsErr bool
 	volWithLessCap    bool
+	volWithZeroCap    bool
 	expectedPVSpec    *pvSpec
 	withSecretRefs    bool
 	expectErr         bool
@@ -1093,6 +1094,32 @@ func TestProvision(t *testing.T) {
 				}
 			},
 		},
+		"provision with size 0": {
+			volOpts: controller.VolumeOptions{
+				PersistentVolumeReclaimPolicy: v1.PersistentVolumeReclaimDelete,
+				PVName:                        "test-name",
+				PVC:                           createFakePVC(requestedBytes),
+				Parameters: map[string]string{
+					"fstype": "ext3",
+				},
+			},
+			volWithZeroCap: true,
+			expectedPVSpec: &pvSpec{
+				Name:          "test-testi",
+				ReclaimPolicy: v1.PersistentVolumeReclaimDelete,
+				Capacity: v1.ResourceList{
+					v1.ResourceName(v1.ResourceStorage): bytesToGiQuantity(0),
+				},
+				CSIPVS: &v1.CSIPersistentVolumeSource{
+					Driver:       "test-driver",
+					VolumeHandle: "test-volume-id",
+					FSType:       "ext3",
+					VolumeAttributes: map[string]string{
+						"storage.kubernetes.io/csiProvisionerIdentity": "test-provisioner",
+					},
+				},
+			},
+		},
 	}
 
 	for k, tc := range testcases {
@@ -1196,6 +1223,9 @@ func runProvisionTest(t *testing.T, k string, tc provisioningTestcase, requested
 		out.Volume.CapacityBytes = int64(80)
 		controllerServer.EXPECT().CreateVolume(gomock.Any(), gomock.Any()).Return(out, nil).Times(1)
 		controllerServer.EXPECT().DeleteVolume(gomock.Any(), gomock.Any()).Return(&csi.DeleteVolumeResponse{}, nil).Times(1)
+	} else if tc.volWithZeroCap {
+		out.Volume.CapacityBytes = int64(0)
+		controllerServer.EXPECT().CreateVolume(gomock.Any(), gomock.Any()).Return(out, nil).Times(1)
 	} else if tc.expectCreateVolDo != nil {
 		controllerServer.EXPECT().CreateVolume(gomock.Any(), gomock.Any()).Do(tc.expectCreateVolDo).Return(out, nil).Times(1)
 	} else {