diff --git a/pkg/volume/csi/csi_attacher.go b/pkg/volume/csi/csi_attacher.go
index e429dc9a645..18c6c0a7899 100644
--- a/pkg/volume/csi/csi_attacher.go
+++ b/pkg/volume/csi/csi_attacher.go
@@ -239,23 +239,6 @@ func (c *csiAttacher) MountDevice(spec *volume.Spec, devicePath string, deviceMo
 		return errors.New(log("attacher.MountDevice failed, deviceMountPath is empty"))
 	}
 
-	corruptedDir := false
-	mounted, err := isDirMounted(c.plugin, deviceMountPath)
-	if err != nil {
-		klog.Error(log("attacher.MountDevice failed while checking mount status for dir [%s]", deviceMountPath))
-		if isCorruptedDir(deviceMountPath) {
-			corruptedDir = true // leave to CSI driver to handle corrupted mount
-			klog.Warning(log("attacher.MountDevice detected corrupted mount for dir [%s]", deviceMountPath))
-		} else {
-			return err
-		}
-	}
-
-	if mounted && !corruptedDir {
-		klog.V(4).Info(log("attacher.MountDevice skipping mount, dir already mounted [%s]", deviceMountPath))
-		return nil
-	}
-
 	// Setup
 	if spec == nil {
 		return errors.New(log("attacher.MountDevice failed, spec is nil"))
@@ -304,7 +287,7 @@ func (c *csiAttacher) MountDevice(spec *volume.Spec, devicePath string, deviceMo
 
 	// Store volume metadata for UnmountDevice. Keep it around even if the
 	// driver does not support NodeStage, UnmountDevice still needs it.
-	if err = os.MkdirAll(deviceMountPath, 0750); err != nil && !corruptedDir {
+	if err = os.MkdirAll(deviceMountPath, 0750); err != nil {
 		return errors.New(log("attacher.MountDevice failed to create dir %#v:  %v", deviceMountPath, err))
 	}
 	klog.V(4).Info(log("created target path successfully [%s]", deviceMountPath))
diff --git a/pkg/volume/csi/csi_attacher_test.go b/pkg/volume/csi/csi_attacher_test.go
index db3396a97f3..80dd0ee3e70 100644
--- a/pkg/volume/csi/csi_attacher_test.go
+++ b/pkg/volume/csi/csi_attacher_test.go
@@ -1257,6 +1257,18 @@ func TestAttacherMountDevice(t *testing.T) {
 					t.Errorf("expected mount options: %v, got: %v", tc.spec.PersistentVolume.Spec.MountOptions, vol.MountFlags)
 				}
 			}
+
+			// Verify the deviceMountPath was created by the plugin
+			if tc.stageUnstageSet {
+				s, err := os.Stat(tc.deviceMountPath)
+				if err != nil {
+					t.Errorf("expected staging directory %s to be created and be a directory, got error: %s", tc.deviceMountPath, err)
+				} else {
+					if !s.IsDir() {
+						t.Errorf("expected staging directory %s to be directory, got something else", tc.deviceMountPath)
+					}
+				}
+			}
 		})
 	}
 }
diff --git a/pkg/volume/csi/csi_mounter.go b/pkg/volume/csi/csi_mounter.go
index b6e63328915..b5600f04b99 100644
--- a/pkg/volume/csi/csi_mounter.go
+++ b/pkg/volume/csi/csi_mounter.go
@@ -107,22 +107,6 @@ func (c *csiMountMgr) SetUp(mounterArgs volume.MounterArgs) error {
 func (c *csiMountMgr) SetUpAt(dir string, mounterArgs volume.MounterArgs) error {
 	klog.V(4).Infof(log("Mounter.SetUpAt(%s)", dir))
 
-	corruptedDir := false
-	mounted, err := isDirMounted(c.plugin, dir)
-	if err != nil {
-		if isCorruptedDir(dir) {
-			corruptedDir = true // leave to CSI driver to handle corrupted mount
-			klog.Warning(log("mounter.SetUpAt detected corrupted mount for dir [%s]", dir))
-		} else {
-			return errors.New(log("mounter.SetUpAt failed while checking mount status for dir [%s]: %v", dir, err))
-		}
-	}
-
-	if mounted && !corruptedDir {
-		klog.V(4).Info(log("mounter.SetUpAt skipping mount, dir already mounted [%s]", dir))
-		return nil
-	}
-
 	csi, err := c.csiClientGetter.Get()
 	if err != nil {
 		return volumetypes.NewTransientOperationFailure(log("mounter.SetUpAt failed to get CSI client: %v", err))
@@ -219,10 +203,11 @@ func (c *csiMountMgr) SetUpAt(dir string, mounterArgs volume.MounterArgs) error
 	}
 
 	// create target_dir before call to NodePublish
-	if err := os.MkdirAll(dir, 0750); err != nil && !corruptedDir {
-		return errors.New(log("mounter.SetUpAt failed to create dir %#v:  %v", dir, err))
+	parentDir := filepath.Dir(dir)
+	if err := os.MkdirAll(parentDir, 0750); err != nil {
+		return errors.New(log("mounter.SetUpAt failed to create dir %#v:  %v", parentDir, err))
 	}
-	klog.V(4).Info(log("created target path successfully [%s]", dir))
+	klog.V(4).Info(log("created target path successfully [%s]", parentDir))
 
 	nodePublishSecrets = map[string]string{}
 	if secretRef != nil {
diff --git a/pkg/volume/util/operationexecutor/operation_generator.go b/pkg/volume/util/operationexecutor/operation_generator.go
index b3e6eb8b8d0..7d51ac80645 100644
--- a/pkg/volume/util/operationexecutor/operation_generator.go
+++ b/pkg/volume/util/operationexecutor/operation_generator.go
@@ -571,7 +571,7 @@ func (og *operationGenerator) GenerateMountVolumeFunc(
 			DevicePath: devicePath,
 		}
 
-		if volumeDeviceMounter != nil {
+		if volumeDeviceMounter != nil && actualStateOfWorld.GetDeviceMountState(volumeToMount.VolumeName) != DeviceGloballyMounted {
 			deviceMountPath, err :=
 				volumeDeviceMounter.GetDeviceMountPath(volumeToMount.VolumeSpec)
 			if err != nil {
@@ -615,6 +615,17 @@ func (og *operationGenerator) GenerateMountVolumeFunc(
 
 			if resizeError != nil {
 				klog.Errorf("MountVolume.NodeExpandVolume failed with %v", resizeError)
+
+				// Resize failed. To make sure NodeExpand is re-tried again on the next attempt
+				// *before* SetUp(), mark the mounted device as uncertain.
+				markDeviceUncertainErr := actualStateOfWorld.MarkDeviceAsUncertain(
+					volumeToMount.VolumeName, devicePath, deviceMountPath)
+				if markDeviceUncertainErr != nil {
+					// just log, return the resizeError error instead
+					klog.Infof(volumeToMount.GenerateMsgDetailed(
+						"MountVolume.MountDevice failed to mark volume as uncertain",
+						markDeviceUncertainErr.Error()))
+				}
 				return volumeToMount.GenerateError("MountVolume.MountDevice failed while expanding volume", resizeError)
 			}
 		}