diff --git a/docs/deploy-cephfs.md b/docs/deploy-cephfs.md
index b5825ee0216e..883dff69e438 100644
--- a/docs/deploy-cephfs.md
+++ b/docs/deploy-cephfs.md
@@ -79,6 +79,7 @@ is used to define in which namespace you want the configmaps to be stored
 | `fsName`                                                                                            | yes            | CephFS filesystem name into which the volume shall be created                                                                                                                                                           |
 | `mounter`                                                                                           | no             | Mount method to be used for this volume. Available options are `kernel` for Ceph kernel client and `fuse` for Ceph FUSE driver. Defaults to "default mounter", see command line arguments.                              |
 | `pool`                                                                                              | no             | Ceph pool into which volume data shall be stored                                                                                                                                                                        |
+| `mountOptions`                                                                                      | no             | Comma seperated string of mount options accepted by cephfs kernel and/or fuse mount. by default no options are passed. Check man mount.ceph for options.                                                                |
 | `csi.storage.k8s.io/provisioner-secret-name`, `csi.storage.k8s.io/node-stage-secret-name`           | for Kubernetes | Name of the Kubernetes Secret object containing Ceph client credentials. Both parameters should have the same value                                                                                                     |
 | `csi.storage.k8s.io/provisioner-secret-namespace`, `csi.storage.k8s.io/node-stage-secret-namespace` | for Kubernetes | Namespaces of the above Secret objects                                                                                                                                                                                  |
 
diff --git a/e2e/cephfs.go b/e2e/cephfs.go
index 9cf78812180a..e261204dee46 100644
--- a/e2e/cephfs.go
+++ b/e2e/cephfs.go
@@ -152,7 +152,7 @@ var _ = Describe("cephfs", func() {
 				})
 
 				By("check data persist after recreating pod with same pvc", func() {
-					err := checkDataPersist(pvcPath, appPath, f)
+					err := checkDataPersist(pvcPath, appPath, f, true)
 					if err != nil {
 						Fail(err.Error())
 					}
diff --git a/e2e/rbd.go b/e2e/rbd.go
index 3b6dba942129..7b5c34354036 100644
--- a/e2e/rbd.go
+++ b/e2e/rbd.go
@@ -219,7 +219,7 @@ var _ = Describe("RBD", func() {
 			})
 
 			By("check data persist after recreating pod with same pvc", func() {
-				err := checkDataPersist(pvcPath, appPath, f)
+				err := checkDataPersist(pvcPath, appPath, f, false)
 				if err != nil {
 					Fail(err.Error())
 				}
diff --git a/e2e/utils.go b/e2e/utils.go
index 0dd409cd8a7f..e7bbf70d96ef 100644
--- a/e2e/utils.go
+++ b/e2e/utils.go
@@ -211,6 +211,8 @@ func createCephfsStorageClass(c kubernetes.Interface, f *framework.Framework, en
 	sc.Parameters["clusterID"] = fsID
 	_, err := c.StorageV1().StorageClasses().Create(&sc)
 	Expect(err).Should(BeNil())
+
+	sc.Parameters["mountOptions"] = "dirstat"
 }
 
 func createRBDStorageClass(c kubernetes.Interface, f *framework.Framework) {
@@ -781,7 +783,7 @@ func listRBDImages(f *framework.Framework) []string {
 // 	return snapInfos, err
 // }
 
-func checkDataPersist(pvcPath, appPath string, f *framework.Framework) error {
+func checkDataPersist(pvcPath, appPath string, f *framework.Framework, mountOpts bool) error {
 	data := "checking data persist"
 	pvc, err := loadPVC(pvcPath)
 	if pvc == nil {
@@ -806,8 +808,10 @@ func checkDataPersist(pvcPath, appPath string, f *framework.Framework) error {
 		LabelSelector: "app=validate-data",
 	}
 	// write data to PVC
-	filePath := app.Spec.Containers[0].VolumeMounts[0].MountPath + "/test"
+	dirPath := app.Spec.Containers[0].VolumeMounts[0].MountPath + "/testdir/"
+	filePath := dirPath + "test"
 
+	execCommandInPod(f, fmt.Sprintf("mkdir -p %s", dirPath), app.Namespace, &opt)
 	execCommandInPod(f, fmt.Sprintf("echo %s > %s", data, filePath), app.Namespace, &opt)
 
 	// delete app
@@ -826,6 +830,14 @@ func checkDataPersist(pvcPath, appPath string, f *framework.Framework) error {
 		return fmt.Errorf("data not persistent expected data %s received data %s  ", data, persistData)
 	}
 
+	if mountOpts {
+		dirstat, stdErr := execCommandInPod(f, fmt.Sprintf("cat %s", dirPath), app.Namespace, &opt)
+		Expect(stdErr).Should(BeEmpty())
+		if !strings.Contains(dirstat, "entries") {
+			return fmt.Errorf("mount options for cephfs mount not working, received data %s ", dirstat)
+		}
+	}
+
 	err = deletePVCAndApp("", f, pvc, app)
 	return err
 }
diff --git a/examples/cephfs/storageclass.yaml b/examples/cephfs/storageclass.yaml
index 2db52a0a62e0..de561213c7bb 100644
--- a/examples/cephfs/storageclass.yaml
+++ b/examples/cephfs/storageclass.yaml
@@ -20,6 +20,10 @@ parameters:
   # (optional) Ceph pool into which volume data shall be stored
   # pool: cephfs_data
 
+  # (optional) Comma seperated string of Cephfs kernel or ceph fuse
+  # mount options. Check man mount.ceph for mount options. For eg:
+  # mountOptions: readdir_max_bytes=1048576,norbytes
+
   # The secrets have to contain user and/or Ceph admin credentials.
   csi.storage.k8s.io/provisioner-secret-name: csi-cephfs-secret
   csi.storage.k8s.io/provisioner-secret-namespace: default
diff --git a/pkg/cephfs/volumemounter.go b/pkg/cephfs/volumemounter.go
index 7b656997cf66..082cc178d6d0 100644
--- a/pkg/cephfs/volumemounter.go
+++ b/pkg/cephfs/volumemounter.go
@@ -129,10 +129,15 @@ func mountFuse(ctx context.Context, mountPoint string, cr *util.Credentials, vol
 		args = append(args, "--client_mds_namespace="+volOptions.FsName)
 	}
 
+	if volOptions.MountOptions != "" {
+		args = append(args, volOptions.MountOptions)
+	}
+
 	_, stderr, err := execCommand(ctx, "ceph-fuse", args[:]...)
 	if err != nil {
 		return err
 	}
+	klog.Infof("Fuse Mounted, command: %v err: %s", args, stderr)
 
 	// Parse the output:
 	// We need "starting fuse" meaning the mount is ok
@@ -183,6 +188,12 @@ func mountKernel(ctx context.Context, mountPoint string, cr *util.Credentials, v
 	}
 	args = append(args, "-o", optionsStr)
 
+	if volOptions.MountOptions != "" {
+		args = append(args, volOptions.MountOptions)
+	}
+
+	klog.Infof("Kernel Mounted, command: %v", args)
+
 	return execCommandErr(ctx, "mount", args[:]...)
 }
 
diff --git a/pkg/cephfs/volumeoptions.go b/pkg/cephfs/volumeoptions.go
index 36d5fd072b33..7afce7fb336d 100644
--- a/pkg/cephfs/volumeoptions.go
+++ b/pkg/cephfs/volumeoptions.go
@@ -38,6 +38,7 @@ type volumeOptions struct {
 	RootPath        string `json:"rootPath"`
 	Mounter         string `json:"mounter"`
 	ProvisionVolume bool   `json:"provisionVolume"`
+	MountOptions    string `json:"mountOptions"`
 }
 
 func validateNonEmptyField(field, fieldName string) error {
@@ -147,6 +148,10 @@ func newVolumeOptions(ctx context.Context, requestName string, size int64, volOp
 		return nil, err
 	}
 
+	if err = extractOptionalOption(&opts.MountOptions, "mountOptions", volOptions); err != nil {
+		return nil, err
+	}
+
 	opts.RequestName = requestName
 	opts.Size = size