diff --git a/pkg/kubelet/volumemanager/volume_manager.go b/pkg/kubelet/volumemanager/volume_manager.go
index e7a962aa67e..460448ded18 100644
--- a/pkg/kubelet/volumemanager/volume_manager.go
+++ b/pkg/kubelet/volumemanager/volume_manager.go
@@ -419,18 +419,21 @@ func (vm *volumeManager) WaitForAttachAndMount(pod *v1.Pod) error {
 	if err != nil {
 		unmountedVolumes :=
 			vm.getUnmountedVolumes(uniquePodName, expectedVolumes)
-		// Also get unattached volumes for error message
+		// Also get unattached volumes and volumes not in dsw for error message
 		unattachedVolumes :=
-			vm.getUnattachedVolumes(expectedVolumes)
+			vm.getUnattachedVolumes(uniquePodName)
+		volumesNotInDSW :=
+			vm.getVolumesNotInDSW(uniquePodName, expectedVolumes)
 
 		if len(unmountedVolumes) == 0 {
 			return nil
 		}
 
 		return fmt.Errorf(
-			"unmounted volumes=%v, unattached volumes=%v: %s",
+			"unmounted volumes=%v, unattached volumes=%v, failed to process volumes=%v: %s",
 			unmountedVolumes,
 			unattachedVolumes,
+			volumesNotInDSW,
 			err)
 	}
 
@@ -474,15 +477,30 @@ func (vm *volumeManager) WaitForUnmount(pod *v1.Pod) error {
 	return nil
 }
 
-// getUnattachedVolumes returns a list of the volumes that are expected to be attached but
-// are not currently attached to the node
-func (vm *volumeManager) getUnattachedVolumes(expectedVolumes []string) []string {
-	unattachedVolumes := []string{}
-	for _, volume := range expectedVolumes {
-		if !vm.actualStateOfWorld.VolumeExists(v1.UniqueVolumeName(volume)) {
-			unattachedVolumes = append(unattachedVolumes, volume)
+func (vm *volumeManager) getVolumesNotInDSW(uniquePodName types.UniquePodName, expectedVolumes []string) []string {
+	volumesNotInDSW := sets.NewString(expectedVolumes...)
+
+	for _, volumeToMount := range vm.desiredStateOfWorld.GetVolumesToMount() {
+		if volumeToMount.PodName == uniquePodName {
+			volumesNotInDSW.Delete(volumeToMount.OuterVolumeSpecName)
 		}
 	}
+
+	return volumesNotInDSW.List()
+}
+
+// getUnattachedVolumes returns a list of the volumes that are expected to be attached but
+// are not currently attached to the node
+func (vm *volumeManager) getUnattachedVolumes(uniquePodName types.UniquePodName) []string {
+	unattachedVolumes := []string{}
+	for _, volumeToMount := range vm.desiredStateOfWorld.GetVolumesToMount() {
+		if volumeToMount.PodName == uniquePodName &&
+			volumeToMount.PluginIsAttachable &&
+			!vm.actualStateOfWorld.VolumeExists(volumeToMount.VolumeName) {
+			unattachedVolumes = append(unattachedVolumes, volumeToMount.OuterVolumeSpecName)
+		}
+	}
+
 	return unattachedVolumes
 }
 
diff --git a/pkg/kubelet/volumemanager/volume_manager_test.go b/pkg/kubelet/volumemanager/volume_manager_test.go
index 0199db171be..a721f9249e0 100644
--- a/pkg/kubelet/volumemanager/volume_manager_test.go
+++ b/pkg/kubelet/volumemanager/volume_manager_test.go
@@ -21,6 +21,7 @@ import (
 	"os"
 	"reflect"
 	"strconv"
+	"strings"
 	"testing"
 	"time"
 
@@ -137,6 +138,82 @@ func TestGetMountedVolumesForPodAndGetVolumesInUse(t *testing.T) {
 	}
 }
 
+func TestWaitForAttachAndMountError(t *testing.T) {
+	tmpDir, err := utiltesting.MkTmpdir("volumeManagerTest")
+	if err != nil {
+		t.Fatalf("can't make a temp dir: %v", err)
+	}
+	defer os.RemoveAll(tmpDir)
+	podManager := kubepod.NewBasicPodManager(podtest.NewFakeMirrorClient())
+
+	pod := &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "abc",
+			Namespace: "nsA",
+			UID:       "1234",
+		},
+		Spec: v1.PodSpec{
+			Containers: []v1.Container{
+				{
+					Name: "container1",
+					VolumeMounts: []v1.VolumeMount{
+						{
+							Name:      volumetest.FailMountDeviceVolumeName,
+							MountPath: "/vol1",
+						},
+						{
+							Name:      "vol2",
+							MountPath: "/vol2",
+						},
+						{
+							Name:      "vol3",
+							MountPath: "/vol3",
+						},
+					},
+				},
+			},
+			Volumes: []v1.Volume{
+				{
+					Name: volumetest.FailMountDeviceVolumeName,
+					VolumeSource: v1.VolumeSource{
+						ConfigMap: &v1.ConfigMapVolumeSource{},
+					},
+				},
+				{
+					Name: "vol2",
+					VolumeSource: v1.VolumeSource{
+						RBD: &v1.RBDVolumeSource{},
+					},
+				},
+				{
+					Name: "vol3",
+					VolumeSource: v1.VolumeSource{
+						AzureDisk: &v1.AzureDiskVolumeSource{},
+					},
+				},
+			},
+		},
+	}
+
+	kubeClient := fake.NewSimpleClientset(pod)
+
+	manager := newTestVolumeManager(t, tmpDir, podManager, kubeClient, nil)
+
+	stopCh := runVolumeManager(manager)
+	defer close(stopCh)
+
+	podManager.SetPods([]*v1.Pod{pod})
+
+	err = manager.WaitForAttachAndMount(pod)
+	if err == nil {
+		t.Errorf("Expected error, got none")
+	}
+	if !strings.Contains(err.Error(),
+		"unattached volumes=[vol2], failed to process volumes=[vol3]") {
+		t.Errorf("Unexpected error info: %v", err)
+	}
+}
+
 func TestInitialPendingVolumesForPodAndGetVolumesInUse(t *testing.T) {
 	tmpDir, err := utiltesting.MkTmpdir("volumeManagerTest")
 	if err != nil {
@@ -282,14 +359,29 @@ func (p *fakePodStateProvider) ShouldPodContainersBeTerminating(uid kubetypes.UI
 }
 
 func newTestVolumeManager(t *testing.T, tmpDir string, podManager kubepod.Manager, kubeClient clientset.Interface, node *v1.Node) VolumeManager {
-	plug := &volumetest.FakeVolumePlugin{PluginName: "fake", Host: nil}
+	attachablePlug := &volumetest.FakeVolumePlugin{
+		PluginName: "fake",
+		Host:       nil,
+		CanSupportFn: func(spec *volume.Spec) bool {
+			return (spec.PersistentVolume != nil && spec.PersistentVolume.Spec.RBD != nil) ||
+				(spec.Volume != nil && spec.Volume.RBD != nil)
+		},
+	}
+	unattachablePlug := &volumetest.FakeVolumePlugin{
+		PluginName:    "unattachable-fake-plugin",
+		Host:          nil,
+		NonAttachable: true,
+		CanSupportFn: func(spec *volume.Spec) bool {
+			return spec.Volume != nil && spec.Volume.ConfigMap != nil
+		},
+	}
 	fakeRecorder := &record.FakeRecorder{}
 	plugMgr := &volume.VolumePluginMgr{}
 	// TODO (#51147) inject mock prober
 	fakeVolumeHost := volumetest.NewFakeKubeletVolumeHost(t, tmpDir, kubeClient, nil)
 	fakeVolumeHost.WithNode(node)
 
-	plugMgr.InitPlugins([]volume.VolumePlugin{plug}, nil /* prober */, fakeVolumeHost)
+	plugMgr.InitPlugins([]volume.VolumePlugin{attachablePlug, unattachablePlug}, nil /* prober */, fakeVolumeHost)
 	stateProvider := &fakePodStateProvider{}
 	fakePathHandler := volumetest.NewBlockVolumePathHandler()
 	vm := NewVolumeManager(
diff --git a/pkg/volume/testing/testing.go b/pkg/volume/testing/testing.go
index ce6a3cc43db..b1040fd8d19 100644
--- a/pkg/volume/testing/testing.go
+++ b/pkg/volume/testing/testing.go
@@ -183,6 +183,7 @@ type FakeVolumePlugin struct {
 	SupportsRemount        bool
 	SupportsSELinux        bool
 	DisableNodeExpansion   bool
+	CanSupportFn           func(*volume.Spec) bool
 
 	// default to false which means it is attachable by default
 	NonAttachable bool
@@ -269,7 +270,10 @@ func (plugin *FakeVolumePlugin) GetVolumeName(spec *volume.Spec) (string, error)
 }
 
 func (plugin *FakeVolumePlugin) CanSupport(spec *volume.Spec) bool {
-	// TODO: maybe pattern-match on spec.Name() to decide?
+	if plugin.CanSupportFn != nil {
+		return plugin.CanSupportFn(spec)
+	}
+
 	return true
 }