diff --git a/test/e2e/pd.go b/test/e2e/pd.go
index 55a7ef23e5d..41de6602d37 100644
--- a/test/e2e/pd.go
+++ b/test/e2e/pd.go
@@ -75,7 +75,7 @@ var _ = framework.KubeDescribe("Pod Disks", func() {
 	})
 
 	// Flaky-- Issue #27691
-	It("[Flaky] should schedule a pod w/ a RW PD, remove it, then schedule it on another host [Slow]", func() {
+	It("[Flaky] should schedule a pod w/ a RW PD, ungracefully remove it, then schedule it on another host [Slow]", func() {
 		framework.SkipUnlessProviderIs("gce", "gke", "aws")
 
 		By("creating PD")
@@ -111,6 +111,7 @@ var _ = framework.KubeDescribe("Pod Disks", func() {
 		framework.ExpectNoError(waitForPDInVolumesInUse(nodeClient, diskName, host0Name, nodeStatusTimeout, true /* shouldExist */))
 
 		By("deleting host0Pod")
+		// Delete pod with 0 grace period
 		framework.ExpectNoError(podClient.Delete(host0Pod.Name, api.NewDeleteOptions(0)), "Failed to delete host0Pod")
 
 		By("submitting host1Pod to kubernetes")
@@ -134,8 +135,68 @@ var _ = framework.KubeDescribe("Pod Disks", func() {
 		return
 	})
 
+	It("Should schedule a pod w/ a RW PD, gracefully remove it, then schedule it on another host [Slow]", func() {
+		framework.SkipUnlessProviderIs("gce", "gke", "aws")
+
+		By("creating PD")
+		diskName, err := createPDWithRetry()
+		framework.ExpectNoError(err, "Error creating PD")
+
+		host0Pod := testPDPod([]string{diskName}, host0Name, false /* readOnly */, 1 /* numContainers */)
+		host1Pod := testPDPod([]string{diskName}, host1Name, false /* readOnly */, 1 /* numContainers */)
+		containerName := "mycontainer"
+
+		defer func() {
+			// Teardown pods, PD. Ignore errors.
+			// Teardown should do nothing unless test failed.
+			By("cleaning up PD-RW test environment")
+			podClient.Delete(host0Pod.Name, &api.DeleteOptions{})
+			podClient.Delete(host1Pod.Name, &api.DeleteOptions{})
+			detachAndDeletePDs(diskName, []string{host0Name, host1Name})
+		}()
+
+		By("submitting host0Pod to kubernetes")
+		_, err = podClient.Create(host0Pod)
+		framework.ExpectNoError(err, fmt.Sprintf("Failed to create host0Pod: %v", err))
+
+		framework.ExpectNoError(f.WaitForPodRunningSlow(host0Pod.Name))
+
+		testFile := "/testpd1/tracker"
+		testFileContents := fmt.Sprintf("%v", mathrand.Int())
+
+		framework.ExpectNoError(f.WriteFileViaContainer(host0Pod.Name, containerName, testFile, testFileContents))
+		framework.Logf("Wrote value: %v", testFileContents)
+
+		// Verify that disk shows up for in node 1's VolumeInUse list
+		framework.ExpectNoError(waitForPDInVolumesInUse(nodeClient, diskName, host0Name, nodeStatusTimeout, true /* shouldExist */))
+
+		By("deleting host0Pod")
+		// Delete pod with default grace period 30s
+		framework.ExpectNoError(podClient.Delete(host0Pod.Name, &api.DeleteOptions{}), "Failed to delete host0Pod")
+
+		By("submitting host1Pod to kubernetes")
+		_, err = podClient.Create(host1Pod)
+		framework.ExpectNoError(err, "Failed to create host1Pod")
+
+		framework.ExpectNoError(f.WaitForPodRunningSlow(host1Pod.Name))
+
+		v, err := f.ReadFileViaContainer(host1Pod.Name, containerName, testFile)
+		framework.ExpectNoError(err)
+		framework.Logf("Read value: %v", v)
+
+		Expect(strings.TrimSpace(v)).To(Equal(strings.TrimSpace(testFileContents)))
+
+		// Verify that disk is removed from node 1's VolumeInUse list
+		framework.ExpectNoError(waitForPDInVolumesInUse(nodeClient, diskName, host0Name, nodeStatusTimeout, false /* shouldExist */))
+
+		By("deleting host1Pod")
+		framework.ExpectNoError(podClient.Delete(host1Pod.Name, &api.DeleteOptions{}), "Failed to delete host1Pod")
+
+		return
+	})
+
 	// Flaky-- Issue #27477
-	It("[Flaky] should schedule a pod w/ a readonly PD on two hosts, then remove both. [Slow]", func() {
+	It("[Flaky] should schedule a pod w/ a readonly PD on two hosts, then remove both ungracefully. [Slow]", func() {
 		framework.SkipUnlessProviderIs("gce", "gke")
 
 		By("creating PD")
@@ -160,6 +221,7 @@ var _ = framework.KubeDescribe("Pod Disks", func() {
 		_, err = podClient.Create(rwPod)
 		framework.ExpectNoError(err, "Failed to create rwPod")
 		framework.ExpectNoError(f.WaitForPodRunningSlow(rwPod.Name))
+		// Delete pod with 0 grace period
 		framework.ExpectNoError(podClient.Delete(rwPod.Name, api.NewDeleteOptions(0)), "Failed to delete host0Pod")
 		framework.ExpectNoError(waitForPDDetach(diskName, host0Name))
 
@@ -182,6 +244,54 @@ var _ = framework.KubeDescribe("Pod Disks", func() {
 		framework.ExpectNoError(podClient.Delete(host1ROPod.Name, api.NewDeleteOptions(0)), "Failed to delete host1ROPod")
 	})
 
+	It("Should schedule a pod w/ a readonly PD on two hosts, then remove both gracefully. [Slow]", func() {
+		framework.SkipUnlessProviderIs("gce", "gke")
+
+		By("creating PD")
+		diskName, err := createPDWithRetry()
+		framework.ExpectNoError(err, "Error creating PD")
+
+		rwPod := testPDPod([]string{diskName}, host0Name, false /* readOnly */, 1 /* numContainers */)
+		host0ROPod := testPDPod([]string{diskName}, host0Name, true /* readOnly */, 1 /* numContainers */)
+		host1ROPod := testPDPod([]string{diskName}, host1Name, true /* readOnly */, 1 /* numContainers */)
+
+		defer func() {
+			By("cleaning up PD-RO test environment")
+			// Teardown pods, PD. Ignore errors.
+			// Teardown should do nothing unless test failed.
+			podClient.Delete(rwPod.Name, &api.DeleteOptions{})
+			podClient.Delete(host0ROPod.Name, &api.DeleteOptions{})
+			podClient.Delete(host1ROPod.Name, &api.DeleteOptions{})
+			detachAndDeletePDs(diskName, []string{host0Name, host1Name})
+		}()
+
+		By("submitting rwPod to ensure PD is formatted")
+		_, err = podClient.Create(rwPod)
+		framework.ExpectNoError(err, "Failed to create rwPod")
+		framework.ExpectNoError(f.WaitForPodRunningSlow(rwPod.Name))
+		// Delete pod with default grace period 30s
+		framework.ExpectNoError(podClient.Delete(rwPod.Name, &api.DeleteOptions{}), "Failed to delete host0Pod")
+		framework.ExpectNoError(waitForPDDetach(diskName, host0Name))
+
+		By("submitting host0ROPod to kubernetes")
+		_, err = podClient.Create(host0ROPod)
+		framework.ExpectNoError(err, "Failed to create host0ROPod")
+
+		By("submitting host1ROPod to kubernetes")
+		_, err = podClient.Create(host1ROPod)
+		framework.ExpectNoError(err, "Failed to create host1ROPod")
+
+		framework.ExpectNoError(f.WaitForPodRunningSlow(host0ROPod.Name))
+
+		framework.ExpectNoError(f.WaitForPodRunningSlow(host1ROPod.Name))
+
+		By("deleting host0ROPod")
+		framework.ExpectNoError(podClient.Delete(host0ROPod.Name, &api.DeleteOptions{}), "Failed to delete host0ROPod")
+
+		By("deleting host1ROPod")
+		framework.ExpectNoError(podClient.Delete(host1ROPod.Name, &api.DeleteOptions{}), "Failed to delete host1ROPod")
+	})
+
 	It("should schedule a pod w/ a RW PD shared between multiple containers, write to PD, delete pod, verify contents, and repeat in rapid succession [Slow]", func() {
 		framework.SkipUnlessProviderIs("gce", "gke", "aws")