diff --git a/builtin/providers/aws/resource_aws_db_instance.go b/builtin/providers/aws/resource_aws_db_instance.go
index 981e87d9a..34db57c05 100644
--- a/builtin/providers/aws/resource_aws_db_instance.go
+++ b/builtin/providers/aws/resource_aws_db_instance.go
@@ -899,6 +899,10 @@ func resourceAwsDbInstanceUpdate(d *schema.ResourceData, meta interface{}) error
 	return resourceAwsDbInstanceRead(d, meta)
 }
 
+// resourceAwsDbInstanceRetrieve fetches DBInstance information from the AWS
+// API. It returns an error if there is a communication problem or unexpected
+// error with AWS. When the DBInstance is not found, it returns no error and a
+// nil pointer.
 func resourceAwsDbInstanceRetrieve(
 	d *schema.ResourceData, meta interface{}) (*rds.DBInstance, error) {
 	conn := meta.(*AWSClient).rdsconn
diff --git a/builtin/providers/aws/resource_aws_rds_cluster_instance.go b/builtin/providers/aws/resource_aws_rds_cluster_instance.go
index 14cbd8ebf..59cf4c57c 100644
--- a/builtin/providers/aws/resource_aws_rds_cluster_instance.go
+++ b/builtin/providers/aws/resource_aws_rds_cluster_instance.go
@@ -123,8 +123,13 @@ func resourceAwsRDSClusterInstanceCreate(d *schema.ResourceData, meta interface{
 
 func resourceAwsRDSClusterInstanceRead(d *schema.ResourceData, meta interface{}) error {
 	db, err := resourceAwsDbInstanceRetrieve(d, meta)
+	// Errors from this helper are always reportable
 	if err != nil {
-		log.Printf("[WARN] Error on retrieving RDS Cluster Instance (%s): %s", d.Id(), err)
+		return fmt.Errorf("[WARN] Error on retrieving RDS Cluster Instance (%s): %s", d.Id(), err)
+	}
+	// A nil response means "not found"
+	if db == nil {
+		log.Printf("[WARN] RDS Cluster Instance (%s): not found, removing from state.", d.Id())
 		d.SetId("")
 		return nil
 	}
diff --git a/builtin/providers/aws/resource_aws_rds_cluster_instance_test.go b/builtin/providers/aws/resource_aws_rds_cluster_instance_test.go
index f9e582672..89f128615 100644
--- a/builtin/providers/aws/resource_aws_rds_cluster_instance_test.go
+++ b/builtin/providers/aws/resource_aws_rds_cluster_instance_test.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/hashicorp/terraform/helper/acctest"
 	"github.com/hashicorp/terraform/helper/resource"
@@ -33,6 +34,28 @@ func TestAccAWSRDSClusterInstance_basic(t *testing.T) {
 	})
 }
 
+// https://github.com/hashicorp/terraform/issues/5350
+func TestAccAWSRDSClusterInstance_disappears(t *testing.T) {
+	var v rds.DBInstance
+
+	resource.Test(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckAWSClusterDestroy,
+		Steps: []resource.TestStep{
+			resource.TestStep{
+				Config: testAccAWSClusterInstanceConfig,
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAWSClusterInstanceExists("aws_rds_cluster_instance.cluster_instances", &v),
+					testAccAWSClusterInstanceDisappears(&v),
+				),
+				// A non-empty plan is what we want. A crash is what we don't want. :)
+				ExpectNonEmptyPlan: true,
+			},
+		},
+	})
+}
+
 func testAccCheckAWSClusterInstanceDestroy(s *terraform.State) error {
 	for _, rs := range s.RootModule().Resources {
 		if rs.Type != "aws_rds_cluster" {
@@ -83,6 +106,34 @@ func testAccCheckAWSDBClusterInstanceAttributes(v *rds.DBInstance) resource.Test
 	}
 }
 
+func testAccAWSClusterInstanceDisappears(v *rds.DBInstance) resource.TestCheckFunc {
+	return func(s *terraform.State) error {
+		conn := testAccProvider.Meta().(*AWSClient).rdsconn
+		opts := &rds.DeleteDBInstanceInput{
+			DBInstanceIdentifier: v.DBInstanceIdentifier,
+		}
+		if _, err := conn.DeleteDBInstance(opts); err != nil {
+			return err
+		}
+		return resource.Retry(40*time.Minute, func() *resource.RetryError {
+			opts := &rds.DescribeDBInstancesInput{
+				DBInstanceIdentifier: v.DBInstanceIdentifier,
+			}
+			_, err := conn.DescribeDBInstances(opts)
+			if err != nil {
+				dbinstanceerr, ok := err.(awserr.Error)
+				if ok && dbinstanceerr.Code() == "DBInstanceNotFound" {
+					return nil
+				}
+				return resource.NonRetryableError(
+					fmt.Errorf("Error retrieving DB Instances: %s", err))
+			}
+			return resource.RetryableError(fmt.Errorf(
+				"Waiting for instance to be deleted: %v", v.DBInstanceIdentifier))
+		})
+	}
+}
+
 func testAccCheckAWSClusterInstanceExists(n string, v *rds.DBInstance) resource.TestCheckFunc {
 	return func(s *terraform.State) error {
 		rs, ok := s.RootModule().Resources[n]