diff --git a/pkg/operator/ceph/cluster/osd/provision_spec.go b/pkg/operator/ceph/cluster/osd/provision_spec.go
index e47668100..eca4f5b04 100644
--- a/pkg/operator/ceph/cluster/osd/provision_spec.go
+++ b/pkg/operator/ceph/cluster/osd/provision_spec.go
@@ -114,7 +114,7 @@ func (c *Cluster) provisionPodTemplateSpec(osdProps osdProperties, restart v1.Re
 
 	if osdProps.onPVC() {
 		// Create volume config for PVCs
-		volumes = append(volumes, getPVCOSDVolumes(&osdProps)...)
+		volumes = append(volumes, getPVCOSDVolumes(&osdProps, c.spec.DataDirHostPath, c.clusterInfo.Namespace, true)...)
 		if osdProps.encrypted {
 			// If a KMS is configured we populate
 			if c.spec.Security.KeyManagementService.IsEnabled() {
diff --git a/pkg/operator/ceph/cluster/osd/spec.go b/pkg/operator/ceph/cluster/osd/spec.go
index 83396cf4b..6bfb52ed5 100644
--- a/pkg/operator/ceph/cluster/osd/spec.go
+++ b/pkg/operator/ceph/cluster/osd/spec.go
@@ -20,7 +20,9 @@ package osd
 import (
 	"fmt"
 	"path"
+	"path/filepath"
 	"strconv"
+	"strings"
 
 	"github.com/libopenstorage/secrets"
 	"github.com/pkg/errors"
@@ -227,6 +229,7 @@ var defaultTuneSlowSettings = map[string]string{
 var cpArgs = []string{
 	"--archive",     // Archive mode preserves the specified attributes
 	"--dereference", // always follow symbolic links in SOURCE to avoid broken links when copying dm devs
+	"--no-clobber",  // do not overwrite an existing file
 	"--verbose",
 }
 
@@ -268,7 +271,7 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo, provisionC
 	// If the OSD runs on PVC
 	if osdProps.onPVC() {
 		// Create volume config for PVCs
-		volumes = append(volumes, getPVCOSDVolumes(&osdProps)...)
+		volumes = append(volumes, getPVCOSDVolumes(&osdProps, c.context.ConfigDir, c.clusterInfo.Namespace, false)...)
 		// If encrypted let's add the secret key mount path
 		if osdProps.encrypted && osd.CVMode == "raw" {
 			encryptedVol, _ := c.getEncryptionVolume(osdProps)
@@ -395,7 +398,7 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo, provisionC
 	// so that it can pick the already mounted/activated osd metadata path
 	// This container will activate the OSD and place the activated filesinto an empty dir
 	// The empty dir will be shared by the "activate-osd" pod and the "osd" main pod
-	activateOSDVolume, activateOSDContainer := c.getActivateOSDInitContainer(osdID, osd, osdProps)
+	activateOSDVolume, activateOSDContainer := c.getActivateOSDInitContainer(c.context.ConfigDir, c.clusterInfo.Namespace, osdID, osd, osdProps)
 	if doActivateOSDInit {
 		volumes = append(volumes, activateOSDVolume)
 		volumeMounts = append(volumeMounts, activateOSDContainer.VolumeMounts[0])
@@ -645,8 +648,23 @@ func (c *Cluster) getCopyBinariesContainer() (v1.Volume, *v1.Container) {
 }
 
 // This container runs all the actions needed to activate an OSD before we can run the OSD process
-func (c *Cluster) getActivateOSDInitContainer(osdID string, osdInfo OSDInfo, osdProps osdProperties) (v1.Volume, *v1.Container) {
-	volume := v1.Volume{Name: activateOSDVolumeName, VolumeSource: v1.VolumeSource{EmptyDir: &v1.EmptyDirVolumeSource{}}}
+func (c *Cluster) getActivateOSDInitContainer(configDir, namespace, osdID string, osdInfo OSDInfo, osdProps osdProperties) (v1.Volume, *v1.Container) {
+	// We need to use hostPath because the same reason as written in the comment of getDataBridgeVolumeSource()
+	hostPathType := v1.HostPathDirectoryOrCreate
+	source := v1.VolumeSource{
+		HostPath: &v1.HostPathVolumeSource{
+			Path: filepath.Join(
+				configDir,
+				namespace,
+				strings.ReplaceAll(osdInfo.BlockPath, "/", "_"),
+			),
+			Type: &hostPathType,
+		},
+	}
+	volume := v1.Volume{
+		Name:         activateOSDVolumeName,
+		VolumeSource: source,
+	}
 	envVars := osdActivateEnvVar()
 	osdStore := "--bluestore"
 
diff --git a/pkg/operator/ceph/cluster/osd/volumes.go b/pkg/operator/ceph/cluster/osd/volumes.go
index d6f88022d..c38a462cb 100644
--- a/pkg/operator/ceph/cluster/osd/volumes.go
+++ b/pkg/operator/ceph/cluster/osd/volumes.go
@@ -19,6 +19,7 @@ package osd
 import (
 	"fmt"
 	"path"
+	"path/filepath"
 
 	"github.com/libopenstorage/secrets"
 	kms "github.com/rook/rook/pkg/daemon/ceph/osd/kms"
@@ -86,7 +87,31 @@ func getDeviceMapperVolume() (v1.Volume, v1.VolumeMount) {
 	return volume, volumeMounts
 }
 
-func getPVCOSDVolumes(osdProps *osdProperties) []v1.Volume {
+func getDataBridgeVolumeSource(claimName, configDir, namespace string, inProvisioning bool) v1.VolumeSource {
+	var source v1.VolumeSource
+	if inProvisioning {
+		source.EmptyDir = &v1.EmptyDirVolumeSource{
+			Medium: "Memory",
+		}
+	} else {
+		// We need to use hostPath to prevent multiple OSD pods from launching the same OSD and causing corruption.
+		// Ceph avoids this problem by locking fsid file and block device file under the data bridge volume directory.
+		// These locks are released by kernel once the process is gone, so until the ceph-osd daemon alives, the other
+		// pods (same OSD) will not be able to acquire them and will continue to be restarted.
+		// If we use emptyDir, this exclusive control doesn't work because the lock files aren't shared between OSD pods.
+		hostPathType := v1.HostPathDirectoryOrCreate
+		source.HostPath = &v1.HostPathVolumeSource{
+			Path: filepath.Join(
+				configDir,
+				namespace,
+				claimName),
+			Type: &hostPathType,
+		}
+	}
+	return source
+}
+
+func getPVCOSDVolumes(osdProps *osdProperties, configDir string, namespace string, prepare bool) []v1.Volume {
 	volumes := []v1.Volume{
 		{
 			Name: osdProps.pvc.ClaimName,
@@ -98,12 +123,8 @@ func getPVCOSDVolumes(osdProps *osdProperties) []v1.Volume {
 			// We need a bridge mount which is basically a common volume mount between the non privileged init container
 			// and the privileged provision container or osd daemon container
 			// The reason for this is mentioned in the comment for getPVCInitContainer() method
-			Name: fmt.Sprintf("%s-bridge", osdProps.pvc.ClaimName),
-			VolumeSource: v1.VolumeSource{
-				EmptyDir: &v1.EmptyDirVolumeSource{
-					Medium: "Memory",
-				},
-			},
+			Name:         fmt.Sprintf("%s-bridge", osdProps.pvc.ClaimName),
+			VolumeSource: getDataBridgeVolumeSource(osdProps.pvc.ClaimName, configDir, namespace, prepare),
 		},
 	}
 
@@ -120,12 +141,8 @@ func getPVCOSDVolumes(osdProps *osdProperties) []v1.Volume {
 				// We need a bridge mount which is basically a common volume mount between the non privileged init container
 				// and the privileged provision container or osd daemon container
 				// The reason for this is mentioned in the comment for getPVCInitContainer() method
-				Name: fmt.Sprintf("%s-bridge", osdProps.metadataPVC.ClaimName),
-				VolumeSource: v1.VolumeSource{
-					EmptyDir: &v1.EmptyDirVolumeSource{
-						Medium: "Memory",
-					},
-				},
+				Name:         fmt.Sprintf("%s-bridge", osdProps.metadataPVC.ClaimName),
+				VolumeSource: getDataBridgeVolumeSource(osdProps.metadataPVC.ClaimName, configDir, namespace, prepare),
 			},
 		}
 
@@ -145,12 +162,8 @@ func getPVCOSDVolumes(osdProps *osdProperties) []v1.Volume {
 				// We need a bridge mount which is basically a common volume mount between the non privileged init container
 				// and the privileged provision container or osd daemon container
 				// The reason for this is mentioned in the comment for getPVCInitContainer() method
-				Name: fmt.Sprintf("%s-bridge", osdProps.walPVC.ClaimName),
-				VolumeSource: v1.VolumeSource{
-					EmptyDir: &v1.EmptyDirVolumeSource{
-						Medium: "Memory",
-					},
-				},
+				Name:         fmt.Sprintf("%s-bridge", osdProps.walPVC.ClaimName),
+				VolumeSource: getDataBridgeVolumeSource(osdProps.walPVC.ClaimName, configDir, namespace, prepare),
 			},
 		}
 
diff --git a/pkg/operator/ceph/cluster/osd/volumes_test.go b/pkg/operator/ceph/cluster/osd/volumes_test.go
index bc267a94f..5f794bccd 100644
--- a/pkg/operator/ceph/cluster/osd/volumes_test.go
+++ b/pkg/operator/ceph/cluster/osd/volumes_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package osd
 
 import (
+	"path/filepath"
 	"testing"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -44,3 +45,15 @@ func TestGetEncryptionVolume(t *testing.T) {
 	assert.Equal(t, v1.Volume{Name: "osd-encryption-key", VolumeSource: v1.VolumeSource{EmptyDir: &v1.EmptyDirVolumeSource{Medium: "Memory"}}}, v)
 	assert.Equal(t, v1.VolumeMount{Name: "osd-encryption-key", ReadOnly: false, MountPath: "/etc/ceph"}, vM)
 }
+
+func TestGetDataBridgeVolumeSource(t *testing.T) {
+	claimName := "test-claim"
+	configDir := "/var/lib/rook"
+	namespace := "rook-ceph"
+
+	source := getDataBridgeVolumeSource(claimName, configDir, namespace, true)
+	assert.Equal(t, v1.VolumeSource{EmptyDir: &v1.EmptyDirVolumeSource{Medium: "Memory"}}, source)
+	hostPathType := v1.HostPathDirectoryOrCreate
+	source = getDataBridgeVolumeSource(claimName, configDir, namespace, false)
+	assert.Equal(t, v1.VolumeSource{HostPath: &v1.HostPathVolumeSource{Path: filepath.Join(configDir, namespace, claimName), Type: &hostPathType}}, source)
+}