diff --git a/examples/persistent-volume-provisioning/README.md b/examples/persistent-volume-provisioning/README.md
index e664ae067d5..4d3f00af648 100644
--- a/examples/persistent-volume-provisioning/README.md
+++ b/examples/persistent-volume-provisioning/README.md
@@ -209,6 +209,7 @@ parameters:
     pool: kube
     userId: kube
     userSecretName: ceph-secret-user
+    imageFormat: "1"
 ```
 
 * `monitors`: Ceph monitors, comma delimited. It is required.
@@ -218,6 +219,10 @@ parameters:
 * `pool`: Ceph RBD pool. Default is "rbd".
 * `userId`: Ceph client ID that is used to map the RBD image. Default is the same as `adminId`.
 * `userSecretName`: The name of Ceph Secret for `userId` to map RBD image. It must exist in the same namespace as PVCs. It is required.
+* `imageFormat`: Ceph RBD image format, "1" or "2". Default is "1".
+* `imageFeatures`: Ceph RBD image format 2 features, comma delimited. This is optional, and only be used if you set `imageFormat` to "2". Currently supported features are `layering` only. Default is "", no features is turned on.
+
+NOTE: We cannot turn on `exclusive-lock` feature for now (and `object-map`, `fast-diff`, `journaling` which require `exclusive-lock`), because exclusive lock and advisory lock cannot work together. (See [#45805](https://issue.k8s.io/45805))
 
 #### Quobyte
 
diff --git a/pkg/volume/rbd/BUILD b/pkg/volume/rbd/BUILD
index f70f9966fe0..90b90d8affc 100644
--- a/pkg/volume/rbd/BUILD
+++ b/pkg/volume/rbd/BUILD
@@ -31,6 +31,7 @@ go_library(
         "//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/uuid:go_default_library",
     ],
 )
diff --git a/pkg/volume/rbd/rbd.go b/pkg/volume/rbd/rbd.go
index 70be1f900a8..92aee629687 100644
--- a/pkg/volume/rbd/rbd.go
+++ b/pkg/volume/rbd/rbd.go
@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/uuid"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
 	"k8s.io/kubernetes/pkg/util/exec"
@@ -35,6 +36,10 @@ import (
 	"k8s.io/kubernetes/pkg/volume/util/volumehelper"
 )
 
+var (
+	supportedFeatures = sets.NewString("layering")
+)
+
 // This is the primary entrypoint for volume plugins.
 func ProbeVolumePlugins() []volume.VolumePlugin {
 	return []volume.VolumePlugin{&rbdPlugin{nil, exec.New()}}
@@ -51,8 +56,10 @@ var _ volume.DeletableVolumePlugin = &rbdPlugin{}
 var _ volume.ProvisionableVolumePlugin = &rbdPlugin{}
 
 const (
-	rbdPluginName = "kubernetes.io/rbd"
-	secretKeyName = "key" // key name used in secret
+	rbdPluginName   = "kubernetes.io/rbd"
+	secretKeyName   = "key" // key name used in secret
+	rbdImageFormat1 = "1"
+	rbdImageFormat2 = "2"
 )
 
 func (plugin *rbdPlugin) Init(host volume.VolumeHost) error {
@@ -267,6 +274,7 @@ func (r *rbdVolumeProvisioner) Provision() (*v1.PersistentVolume, error) {
 	adminSecretNamespace := "default"
 	secretName := ""
 	secret := ""
+	imageFormat := rbdImageFormat1
 
 	for k, v := range r.options.Parameters {
 		switch dstrings.ToLower(k) {
@@ -287,11 +295,27 @@ func (r *rbdVolumeProvisioner) Provision() (*v1.PersistentVolume, error) {
 			r.Pool = v
 		case "usersecretname":
 			secretName = v
+		case "imageformat":
+			imageFormat = v
+		case "imagefeatures":
+			arr := dstrings.Split(v, ",")
+			for _, f := range arr {
+				if !supportedFeatures.Has(f) {
+					return nil, fmt.Errorf("invalid feature %q for volume plugin %s, supported features are: %v", f, r.plugin.GetPluginName(), supportedFeatures)
+				} else {
+					r.imageFeatures = append(r.imageFeatures, f)
+				}
+			}
 		default:
 			return nil, fmt.Errorf("invalid option %q for volume plugin %s", k, r.plugin.GetPluginName())
 		}
 	}
 	// sanity check
+	if imageFormat != rbdImageFormat1 && imageFormat != rbdImageFormat2 {
+		return nil, fmt.Errorf("invalid ceph imageformat %s, expecting %s or %s",
+			imageFormat, rbdImageFormat1, rbdImageFormat2)
+	}
+	r.imageFormat = imageFormat
 	if adminSecretName == "" {
 		return nil, fmt.Errorf("missing Ceph admin secret name")
 	}
@@ -376,14 +400,16 @@ func (rbd *rbd) GetPath() string {
 type rbdMounter struct {
 	*rbd
 	// capitalized so they can be exported in persistRBD()
-	Mon          []string
-	Id           string
-	Keyring      string
-	Secret       string
-	fsType       string
-	adminSecret  string
-	adminId      string
-	mountOptions []string
+	Mon           []string
+	Id            string
+	Keyring       string
+	Secret        string
+	fsType        string
+	adminSecret   string
+	adminId       string
+	mountOptions  []string
+	imageFormat   string
+	imageFeatures []string
 }
 
 var _ volume.Mounter = &rbdMounter{}
diff --git a/pkg/volume/rbd/rbd_util.go b/pkg/volume/rbd/rbd_util.go
index d3deab54fef..3ad12361752 100644
--- a/pkg/volume/rbd/rbd_util.go
+++ b/pkg/volume/rbd/rbd_util.go
@@ -355,9 +355,19 @@ func (util *RBDUtil) CreateImage(p *rbdVolumeProvisioner) (r *v1.RBDVolumeSource
 	// iterate all monitors until create succeeds.
 	for i := start; i < start+l; i++ {
 		mon := p.Mon[i%l]
-		glog.V(4).Infof("rbd: create %s size %s using mon %s, pool %s id %s key %s", p.rbdMounter.Image, volSz, mon, p.rbdMounter.Pool, p.rbdMounter.adminId, p.rbdMounter.adminSecret)
-		output, err = p.rbdMounter.plugin.execCommand("rbd",
-			[]string{"create", p.rbdMounter.Image, "--size", volSz, "--pool", p.rbdMounter.Pool, "--id", p.rbdMounter.adminId, "-m", mon, "--key=" + p.rbdMounter.adminSecret, "--image-format", "1"})
+		if p.rbdMounter.imageFormat == rbdImageFormat2 {
+			glog.V(4).Infof("rbd: create %s size %s format %s (features: %s) using mon %s, pool %s id %s key %s", p.rbdMounter.Image, volSz, p.rbdMounter.imageFormat, p.rbdMounter.imageFeatures, mon, p.rbdMounter.Pool, p.rbdMounter.adminId, p.rbdMounter.adminSecret)
+		} else {
+			glog.V(4).Infof("rbd: create %s size %s format %s using mon %s, pool %s id %s key %s", p.rbdMounter.Image, volSz, p.rbdMounter.imageFormat, mon, p.rbdMounter.Pool, p.rbdMounter.adminId, p.rbdMounter.adminSecret)
+		}
+		args := []string{"create", p.rbdMounter.Image, "--size", volSz, "--pool", p.rbdMounter.Pool, "--id", p.rbdMounter.adminId, "-m", mon, "--key=" + p.rbdMounter.adminSecret, "--image-format", p.rbdMounter.imageFormat}
+		if p.rbdMounter.imageFormat == rbdImageFormat2 {
+			// if no image features is provided, it results in empty string
+			// which disable all RBD image format 2 features as we expected
+			features := strings.Join(p.rbdMounter.imageFeatures, ",")
+			args = append(args, "--image-feature", features)
+		}
+		output, err = p.rbdMounter.plugin.execCommand("rbd", args)
 		if err == nil {
 			break
 		} else {