diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
index 20736e0131..0ee4456d38 100644
--- a/pkg/controller/controller.go
+++ b/pkg/controller/controller.go
@@ -527,7 +527,12 @@ func (p *csiProvisioner) Provision(options controller.ProvisionOptions) (*v1.Per
 		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 {
+		respCap = volSizeBytes
+		klog.V(3).Infof("csiClient response volume with size 0, which is not supported by apiServer, will use claim size:%d", respCap)
+	} 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 84b2099d5d..793bba1d0e 100644
--- a/pkg/controller/controller_test.go
+++ b/pkg/controller/controller_test.go
@@ -764,6 +764,7 @@ type provisioningTestcase struct {
 	getSecretRefErr   bool
 	getCredentialsErr bool
 	volWithLessCap    bool
+	volWithZeroCap    bool
 	expectedPVSpec    *pvSpec
 	withSecretRefs    bool
 	expectErr         bool
@@ -1262,6 +1263,34 @@ func TestProvision(t *testing.T) {
 				}
 			},
 		},
+		"provision with size 0": {
+			volOpts: controller.ProvisionOptions{
+				StorageClass: &storagev1.StorageClass{
+					Parameters: map[string]string{
+						"fstype": "ext3",
+					},
+					ReclaimPolicy: &deletePolicy,
+				},
+				PVName: "test-name",
+				PVC:    createFakePVC(requestedBytes),
+			},
+			volWithZeroCap: true,
+			expectedPVSpec: &pvSpec{
+				Name:          "test-testi",
+				ReclaimPolicy: v1.PersistentVolumeReclaimDelete,
+				Capacity: v1.ResourceList{
+					v1.ResourceName(v1.ResourceStorage): bytesToGiQuantity(requestedBytes),
+				},
+				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 {
@@ -1372,6 +1401,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 {